使用静态文件常用于存放css、html、JavaScript注意是在FastAPI()实例上挂载静态文件路径，不是在APIRouter()路由实例上fromfastapiimportFastAPIfromfastapi.staticfilesimportStaticFilesimportuvicornapp=FastAPI()#注意是在FastAPI()实例上挂载静态文件路径app.mount(path='/static',#网页的路径app=StaticFiles(directory='./static'),#静态文件目录的路径name='static')if__name__=='__main__':uvicorn.run(app)
File上传小文件fromtypingimportListfromfastapiimportFastAPI,Fileimportuvicornapp=FastAPI()@app.post('/file')asyncdefup_file(file:bytes=File(...),#files:List[bytes]=File(...)#一次性上传多个小文件):"""会把整个文件加载到内存，可能会出现爆内存的情况只能上传小文件，能获取字节长度，文件的其他属性不能获取到:paramfile::return:"""return{'file_length':len(file)}if__name__=='__main__':uvicorn.run(app)UpLoadFile上传大文件fromtypingimportListfromfastapiimportFastAPI,File,UploadFileimportuvicornapp=FastAPI()@app.post('/up_file')asyncdefup_file(file:UploadFile=File(...),#files:List[UploadFile]=File(...)#一次性上传多个文件):"""UploadFile的优势：-文件存储在内存，达到阈值后，将保存到磁盘中-适用于图片、视频大文件...-可以获取上传文件的元数据，如文件名、类型、创建时间...-有文件对象的异步接口-上传的文件是Python文件对象，可以使用write()、read()、seed()、close()进行操作"""file_content=awaitfile.read()#读取文件print(file_content)return{'file_name':file.filename,'file_type':file.content_type,}if__name__=='__main__':uvicorn.run(app)
From表单需要安装pipinstallpython-multipartFrom有Body（及Query、Path、Cookie）相同的元数据和验证。fromfastapiimportFastAPI,Formimportuvicornapp=FastAPI()@app.post('/login')asyncdeflogin(user_name:str=Form(...),password:str=Form(...)):return{'user_name':user_name,'password':password}if__name__=='__main__':uvicorn.run(app)
说明作用：响应模型数据格式的规范定义响应体模型fromfastapiimportFastAPIfrompydanticimportBaseModel,EmailStr#pipinstallpydantic[email]importuvicornapp=FastAPI()classResIn(BaseModel):#请求体模型id:intpassword:stremail:EmailStr=Noneqq_num:str='876545500'classResOut(BaseModel):#响应体模型没有password字段id:intemail:EmailStr=Noneqq_num:str='876545500'@app.post('/response',response_model=ResOut,response_model_exclude_unset=True#模型设置的默认值不显示在响应模型中，响应仅仅包含实际的值，如果实际的值与默认值相同也会显示)asyncdefresponse(res_in:ResIn):res=dict()res["id"]=res_in.idres["password"]=res_in.password#虽然赋值密码，实际响应中也不会显示res["email"]=res_in.emailres["qq_num"]=res_in.qq_numreturnresif__name__=='__main__':uvicorn.run(app,host='127.0.0.1',port=8002)请求&响应#请求POSThttp://127.0.0.1:8002/response{"id":0,"password":"xxxxxxxx","email":"user@example.com","qq_num":"876545500"}#响应没有passwors字段{"id":0,"email":"user@example.com","qq_num":"876545500"}响应模型的其他参数fromtypingimportList,Union@app.post('/response',#response_model=ResOut,response_model=Union[ResOut,ResIn],#ResOut&ResIn模型字段的并集（所有字段都有）#response_model=List[ResOut],#响应为包含ResOut模型的列表#response_model_include=['id','qq_num'],#响应必须包含的字段#response_model_exclude=['password'],#响应排除的字段response_model_exclude_unset=True#模型设置的默认值不显示在响应模型中，响应仅仅包含实际的值，如果实际的值与默认值相同也会显示)asyncdefresponse(res_in:ResIn):returnres_in
Cookie参数这里说的Cookie是只在Cookie里携带的值，不是在Headers里面的Cookie字段。Cookie定义fromfastapiimportFastAPI,CookiefromtypingimportOptionalimportuvicornapp=FastAPI()@app.get('/cookie')asyncdefget_cookie(#实例化Cookie类，表示该参数在Cookie里面cookie_id:Optional[str]=Cookie(None)):returncookie_idif__name__=='__main__':uvicorn.run(app='main:app',host='127.0.0.1',port=9000,reload=True,workers=2)Headers参数Headers里面的参数fromfastapiimportFastAPI,HeaderfromtypingimportOptionalimportuvicornapp=FastAPI()@app.get('/header')asyncdefget_cookie(#实例化Header类，表示该参数在头部里面user_agent:Optional[str]=Header(None,convert_underscores=True)#convert_underscores参数表示把头部里的user-agent转变成user_agent):returnuser_agentif__name__=='__main__':uvicorn.run(app='main:app',host='127.0.0.1',port=9000,reload=True,workers=2)
说明pydantic库是python中用于数据接口定义检查与设置管理的库。pydantic在运行时强制执行类型提示，并在数据无效时提供友好的错误。安装：pipinstallpydanticBaseModel基本使用frompydanticimportBaseModelclassInfo(BaseModel):id:intname:strif__name__=='__main__':#实例化使用方式info={'id':1,'name':'Bob'}print(Info(**info))print(Info(id='1',name='Bob'))print(Info(id=1,name='Bob').id)print(Info(id=1,name='Bob').name)print(Info(id=1,name='Bob').json())print(Info(id=1,name='Bob').dict())print(Info(id=1,name='Bob').copy())#浅拷贝>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>id=1name='Bob'id=1name='Bob'1Bob{"id":1,"name":"Bob"}{'id':1,'name':'Bob'}id=1name='Bob'BaseModel错误提示错误提示很详细frompydanticimportBaseModel,ValidationErrorclassInfo(BaseModel):id:intname:strif__name__=='__main__':try:print(Info(id=1,name=[12,34]))exceptValidationErrorase:print(e.json())>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>#提示很详细[{"loc":["name"],"msg":"strtypeexpected","type":"type_error.str"}]BaseModel默认验证类型其他类型：https://pydantic-docs.helpmanual.io/usage/types/相关限制conlist：item_type:Type[T]:列表项的类型min_items:int=None:列表中的最小项目数max_items:int=None:列表中的最大项目数conset：item_type:Type[T]:列表项的类型min_items:int=None:集合中的最小项目数max_items:int=None:集合中的最大项目数conint：strict:bool=False:控制类型强制gt:int=None:强制整数大于设定值ge:int=None:强制整数大于或等于设定值lt:int=None:强制整数小于设定值le:int=None:强制整数小于或等于设定值multiple_of:int=None:强制整数为设定值的倍数confloat：strict:bool=False:控制类型强制gt:int=None:强制浮点数大于设定值ge:int=None:强制浮点数大于或等于设定值lt:int=None:强制浮点数小于设定值le:int=None:强制浮点数小于或等于设定值multiple_of:int=None:强制浮点数为设定值的倍数constr：strip_whitespace:bool=False:删除前尾空格to_lower:bool=False:将所有字符转为小写strict:bool=False:控制类型强制min_length:int=None:字符串的最小长度max_length:int=None:字符串的最大长度regex:str=None:正则表达式来验证字符串frompydanticimportBaseModel,constr,conintfromtypingimportListfromdatetimeimportdateclassInfo(BaseModel):id:int#整形name:str#字符串age:conint(gt=0,le=100)#gt>、ge>=、lt<、le<=time:dateis_boy:bool#布尔friend:List[str]=None#有默认值（此参数可选）自定义组合hobby:List[constr(max_length=255)]#现在str长度if__name__=='__main__':print(Info(id=1,name='Bob',age=12,time="2001-12-23",is_boy=True,friend=['A','B'],hobby=['CC','DD']).json())>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>{"id":1,"name":"Bob","age":12,"time":"2001-12-23","is_boy":true,"friend":["A","B"],"hobby":["CC","DD"]}BaseModel子类嵌套frompydanticimportBaseModel,constr,conintfromtypingimportOptionalclassCity(BaseModel):city_name:strdescription:constr(max_length=255)classInfo(BaseModel):id:int#整形name:str#字符串age:Optional[conint(gt=0,le=100)]#可选gt>、ge>=、lt<、le<=city:Cityif__name__=='__main__':print(Info(id=1,name='Bob',age=12,city={'city_name':'成都','description':'好地方'})..dict())#json中文会被编码>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>#{"id":1,"name":"Bob","age":12,"city":{"city_name":"\u6210\u90fd","description":"\u597d\u5730\u65b9"}}{'id':1,'name':'Bob','age':12,'city':{'city_name':'成都','description':'好地方'}}从ORM对象模型创建BaseModel实例这的ORM对象模型使用的SQLAlchemy模块安装：pip3instaillSQLAlchemy创建ORM对象模型fromsqlalchemyimportColumn,Integer,Stringfromsqlalchemy.ext.declarativeimportdeclarative_baseBase=declarative_base()classORMInfo(Base):#创建ORM映射__tablename__='orm_info'id=Column(Integer,primary_key=True,index=True,autoincrement=True)name=Column(String(100),unique=True,nullable=False)city=Column(String(100),unique=True,nullable=False)frompydanticimportBaseModel,constrclassModelInfo(BaseModel):#创建与ORM对应的BaseModel类id:intname:constr(max_length=100)city:constr(max_length=100)classConfig:orm_mode=True#默认False，需要从ORM实例化数据需要设置Trueorm_info=ORMInfo(id=1,name='Bob',city='zx')print(orm_info)model_info=ModelInfo.from_orm(orm_info)#从ORM模型实例化数据print(model_info)
说明现在网上关于FastAPI的项目结构的资料比较少，以下仅供参考，如有更好的想法，可私聊博主交流项目下载：https://pan.bigdataboy.cn/#/s/oWsb结构说明：项目结构的静态文件使用Jinja2渲染项目├──modules（功能模块）│├──│└──init.py├──routers(路由控制)│├──index.py│├──items.py│├──users.py│└──init.py├──static(静态文件)│├──css│└──js├──templates(网页文件)│└──index.html├──init.py├──app.py(主文件)├──public.py(公共方法)└──config.py(配置文件)结构说明routers(路由控制)控制路由的响应(静态文件，还是Json,…)响应静态文件html.pyfromfastapiimportAPIRouter,HTTPExceptionfromstarlette.templatingimportJinja2Templatesfromstarlette.requestsimportRequestrouter=APIRouter()#绑定模板路径template=Jinja2Templates(directory="templates")@router.get("/")asyncdefindex(request:Request):#request为必传，后面的就跟Jinja2一样returntemplate.TemplateResponse('index.html',{'request':request,"user":"啦啦啦啦"})接口响应路由users.py。/users/路径在本文件定义fromfastapiimportAPIRouterrouter=APIRouter()@router.get("/users/",tags=["users"])asyncdefread_users():return[{"username":"Foo"},{"username":"Bar"}]@router.get("/users/me",tags=["users"])asyncdefread_user_me():return{"username":"fakecurrentuser"}@router.get("/users/{username}",tags=["users"])asyncdefread_user(username:str):return{"username":username}接口响应路由items.py。/items/路径在app.py定义fromfastapiimportAPIRouter,HTTPExceptionrouter=APIRouter()@router.get("/")asyncdefuse_api_munber():return[{"name":"ItemFoo"},{"name":"itemBar"}]@router.get("/{item_id}")asyncdefread_item(item_id:str):return{"name":"FakeSpecificItem","item_id":item_id}@router.put("/{item_id}",tags=["custom"],responses={403:{"description":"Operationforbidden"}},)asyncdefupdate_item(item_id:str):ifitem_id!="foo":raiseHTTPException(status_code=403,detail="Youcanonlyupdatetheitem:foo")return{"item_id":item_id,"name":"TheFighters"}app.py(主文件)fromfastapiimportDepends,FastAPIfromFastAPI_Project.routersimporthtml,items,usersfrompublicimportcount_usefromfastapi.staticfilesimportStaticFilesimportconfigimportuvicornapp=FastAPI(debug=True,version=config.VERSION,)#静态文件绑定app.mount('/static',StaticFiles(directory='static'),name='static')#静态文件路由app.include_router(html.router)#API路由app.include_router(items.router,prefix="/items",#添加路由路径tags=["items"],#标签路由分组作用#注入依赖请求这个路由之前,这个依赖的方法可以用于验证统计等dependencies=[Depends(count_use)],)app.include_router(users.router,dependencies=[Depends(count_use)],responses={404:{"description":"Notfound"}},)if__name__=='__main__':uvicorn.run(app=app,host=config.HOST,port=config.PORT,)config.py配置文件配置参数，链接数据什么的importpymongo#通用配置classConfig:#MongoDB连接地址MONGODB=pymongo.MongoClient("mongodb://localhost:27017/")DATABASES="bigdataboy"pass#运行地址HOST="127.0.0.1"#运行端口PORT=8686#版本VERSION="1.0.1"public.py(公共方法)fromfastapi.loggerimportloggerfromfastapi.requestsimportRequestfromtimeimporttimefromFastAPI_Project.configimportConfig"""记录接口使用信息"""asyncdefcount_use(r:Request):#获取api的路径path=r.url.path.strip("/").split("/")[0]#获取客户端ipip=r.client.host#获取客户端UAuser_agent=r.headers.getlist("user-agent")[-1]logger.info(msg=f"{path}{ip}{user_agent}")#插入mongodbConfig.MONGODB[Config.DATABASES]['count_use_api'].insert({"path":path,#接口路径"ip":ip,#客户端ip"user_agent":user_agent,#客户端UA"timestamp":time()#时间戳})
前言这个是2.0版本，使用现在流行的前后端分离思想重构。体验网址：https://douyin.bigdataboy.cn源码地址：https://github.com/bigdataboy2020/fastapi_douyin使用到的技术后端：语言：PythonWEB框架：FastAPI（现代、快速（高性能）的Web框架）服务器框架：Uvicorn（基于asyncio开发的一个轻量级高效的web服务器框架）反向代理：Nginx(高性能的HTTP和反向代理web服务器)进程管理：Supervisor(ython开发的一套通用的进程管理程序)前端：UI框架：LayUI静态文件存放：Nginx前端部署宝塔添加网站上传前端源码api请求域名可以部署后端之后，再来修改部署后端再添加一个网站，用于添加反向代理修改后端配置&上传源码允许浏览器跨域请求的网址。需要修改成自己的前端域名。宝塔添加Python项目Python版本需要3.6以上。添加进程守护运行目录：/www/wwwroot/api.bigdataboy.cn网站目录启动参数：/www/wwwroot/api.bigdataboy.cn/api_venv/bin/uvicornmain:app按照自己的网站目录改改守护进程配置成功如下，这时只需要去Python项目管理器再启动一次该项目就行。配置反向代理即可部署成功，有问题联系QQ876545500
说明在许多情况下，您需要将错误通知给正在使用API的客户端。FastAPI使用Python的HTTPException异常进行处理，所以raise它。实例fromfastapiimportFastAPI,HTTPExceptionimportuvicornapp=FastAPI()@app.get("/user")asyncdefread_item(age:int,name:str):ifage<18:#返回客户端500状态码及错误原因raiseHTTPException(status_code=500,detail="age小于18")return{"name":name,"age":age}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)添加自定义Headers@app.get("/user")asyncdefread_item(age:int,name:str):ifage<18:#返回客户端500状态码及错误原因raiseHTTPException(status_code=500,#detail可以是list、dict、strdetail={"msg":"age小于18"},#自定义headersheaders={"X-Error":"Theregoesmyerror"})return{"name":name,"age":age}自定义异常处理程序fromfastapiimportFastAPI,Requestfromfastapi.responsesimportJSONResponseimportuvicornapp=FastAPI()#异常参数类classUnicornException(Exception):def__init__(self,status_code:int,msg:str):#错误状态码self.status_code=status_code#错误内容self.msg=msg#异常响应@app.exception_handler(UnicornException)asyncdefunicorn_exception_handler(request:Request,exc:UnicornException):returnJSONResponse(#错误响应状态码status_code=exc.status_code,#错误返回内容格式content={"msg":exc.msg},)@app.get("/user/{name}")asyncdefread_item(name:str):ifname=="bigdata":#错误提示raiseUnicornException(status_code=404,msg="用户名错误，没有该用户")return{"name":name}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)测试URL：http://127.0.0.1/user/bigdata?name=bigdataboy
实例代码fromfastapiimportFastAPIimportuvicornapp=FastAPI()@app.post("/user/{name}")#路径参数asyncdefread_item(name:str):return{"name":name}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)导入验证fromfastapiimportFastAPI,Query可选参数name可传可不传#正常写法name为可选参数@app.get("/user/")asyncdefread_item(age:int,name:str=None):return{"name":name,"age":age}#验证写法name为可选参数@app.get("/user/")asyncdefread_item(age:int,name:str=Query(None)):return{"name":name,"age":age}必选参数name必须传#正常写法name为必选参数asyncdefread_item(age:int,name:str):return{"name":name,"age":age}#验证写法name为必选参数asyncdefread_item(age:int,name:str=Query(...)):return{"name":name,"age":age}默认参数值#正常写法name默认值asyncdefread_item(age:int,name:str="bigdataboy"):return{"name":name,"age":age}#验证写法name默认值asyncdefread_item(age:int,name:str=Query("bigdataboy")):return{"name":name,"age":age}Query验证Query主要支持一下验证长度验证：max_length、min_length正则验证：regex@app.get("/user")#路径参数asyncdefread_item(age:int,name:str=Query(...,#该参数必选min_length=1,#最小长度为1max_length=50,#最大长度为50#regex=re.compile(".+bigdataboy.+").pattern#使用编译的正则表达式需要导入re库regex=".+bigdataboy.+"#正则表达式匹配验证)):return{"name":name,"age":age}其他参数这些参数主要是对接口进行描述@app.get("/user")#路径参数asyncdefread_item(age:int,name:str=Query(...,#该参数必选title="name",#标题description="用户的新名字",#参数作用的描述alias="new-name",#接口参数别名，URL参数就使用该别名deprecated=True,#代表该参数，即将弃用)):return{"name":name,"age":age}
POST请求需要首先定义参数模型fromfastapiimportFastAPIfrompydanticimportBaseModelimportuvicorn#声明参数模型classItem(BaseModel):name:strdescription:str=Noneapp=FastAPI()#接受POST类型@app.post("/")asyncdefread_item(item:Item):#参数类型是returnitemif__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)参数模型#导入模型类frompydanticimportBaseModelclassItem(BaseModel):#必要参数name:str#可选参数description:str=None参看文档http://127.0.0.1/docs请求测试importrequestsurl="http://127.0.0.1"data={"name":"bigdataboy","description":"",}r=requests.post(url=url,json=data)print(r.json())高级用法与路径参数一起使用fromfastapiimportFastAPIfrompydanticimportBaseModelimportuvicorn#声明参数类classItem(BaseModel):name:strdescription:str=Noneapp=FastAPI()@app.post("/user/{id}")#路径参数asyncdefread_item(id:int,item:Item):return{"id":id,**item.dict()}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)
路径参数比如访问：127.0.0.1/user/123/fromfastapiimportFastAPIimportuvicornapp=FastAPI()@app.get("/user/{id}/")asyncdefmain(id:int):return{"id":id}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)API文档地址：http://127.0.0.1/docs固定的路径参数@app.get("/user/id/")asyncdefmain():return{"id":"thecurrentuser"}@app.get("/user/{id}/")asyncdefmain(id:int):return{"id":id}预留路径值传入的路径参数只能在预留路径里，不然就报错fromfastapiimportFastAPI#导入枚举类fromenumimportEnumimportuvicorn#预留路径类继承str，文档将能够知道这些值的类型classModelName(str,Enum):#预留的路径alexnet="alexnet"resnet="resnet"lenet="lenet"app=FastAPI()@app.get("/model/{model_name}")asyncdefget_model(model_name:ModelName):#参数类型是预留路径类#第一种判定写法ifmodel_name==ModelName.alexnet:return{"model_name":model_name,"message":"DeepLearningFTW!"}#第二种判定写法ifmodel_name.value=="lenet":return{"model_name":model_name,"message":"LeCNNalltheimages"}#默认返回return{"model_name":model_name,"message":"Havesomeresiduals"}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)路径转换器访问/files/bigdataboy/a.txt，file_path将能得到bigdataboy/a.txt#添加上:path即可@app.get("/files/{file_path:path}")asyncdefread_file(file_path:str):return{"file_path":file_path}查询参数查询参数是指?后面的&分割的键值对http://127.0.0.1:8000/items/?skip=0&limit=2fromfastapiimportFastAPIimportuvicornapp=FastAPI()fake_items_db=[{"item_name":"Foo"},{"item_name":"Bar"},{"item_name":"Baz"}]@app.get("/items/")#设置的默认值，当默认值是None时，表示该值为可选asyncdefread_item(skip:int=0,limit:int=2,default:str=None):#一个列表切片returnfake_items_db[skip:skip+limit]if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)注意如果声明可选时出现以下类似错误#声明可选limit:int=None#出现以下类似错误Incompatibletypesinassignment(expressionhastype"None",variablehastype"int")这是你需要这样做fromtypingimportOptionallimit:Optional[int]=None完整实例fromfastapiimportFastAPIfromtypingimportOptionalimportuvicornapp=FastAPI()@app.get("/items/")asyncdefread_item(skip:str,limit:Optional[int]=None):item={"skip":skip,"limit":limit}returnitemif__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)