说明有时候我们并不需要依赖注入的返回值，这时我们就可以从路径操作注入依赖路径操作装饰器依赖注入asyncdefverify_ua(user_agent:str=Header(None,convert_underscores=True)):print(f'访问UA：{user_agent}')@app.get('/verify_token',dependencies=[Depends(verify_ua)])asyncdeffun():return'没有返回值'全局依赖注入在FastAPI()、APIRouter()实例化里面使用，这种也是不能获取依赖返回值的asyncdefverify_ua(user_agent:str=Header(None,convert_underscores=True)):print(f'访问UA：{user_agent}')#全局依赖app=FastAPI(dependencies=[Depends(verify_ua)])#路由依赖api=APIRouter(dependencies=[Depends(verify_ua)])
说明FastAPI支持子依赖，就是能按照层级调用依赖，可以用来拆分业务逻辑子依赖fromfastapiimportFastAPI,Dependsfromfastapi.exceptionsimportHTTPExceptionfromtypingimportOptionalimportuvicornapp=FastAPI()#子依赖验证tokenasyncdefcheck_token(token:str):iftoken!='123456':returnFalsereturnTrue#接口依赖asyncdefcheck_token_key(token=Depends(check_token),key:Optional[str]=None):ifnottoken:raiseHTTPException(detail='tokenerror',status_code=400)returnkey@app.get('/sub_depends')asyncdeffun(key=Depends(check_token_key)):returnkeyif__name__=='__main__':uvicorn.run(app)执行流程优化use_cache参数use_cache=True表示多个依赖有一个共同的子依赖时，每次请求只会调用依赖一次，然后从缓存中获取@app.get('/sub_depends')asyncdeffun(key=Depends(check_token_key,use_cache=True)):returnkey
什么是依赖注入依赖注入常用于以下场景共享业务逻辑（复用相同的代码逻辑）共享数据库连接实现安全、验证、角色权限等……上述场景均可以使用依赖注入，将代码重复最小化。依赖注入的作用复用相同代码、简化代码初步使用fromfastapiimportFastAPI,DependsfromtypingimportOptionalimportuvicornapp=FastAPI()#定义一个依赖asyncdefcommon_fun(page:int=1,limit:Optional[int]=10):return{'page':page,'limit':limit,'msg':'依赖被调用'}@app.get('/books')asyncdefget_books(common:dict=Depends(common_fun)):#添加依赖returncommonif__name__=='__main__':uvicorn.run(app)将类作为依赖使用类作为依赖注入，在使用参数的效果上，会发现与模型类一样，所以一般是在使用框架提供的依赖类时使用frompydanticimportBaseModelclassQueryBookModel(BaseModel):page:int=1limit:Optional[int]=3fromfastapiimportFastAPI,DependsfromtypingimportOptionalimportuvicornapp=FastAPI()fake_books_db=[{'book_name':'遥远的救世主'},{'book_name':'天幕红尘'},{'book_name':'背叛'},]#定义一个类classQueryBook(object):def__init__(self,page:int=1,limit:Optional[int]=3):self.page=pageself.limit=limit@app.get('/books')asyncdefget_books(#三种写法common:QueryBook=Depends(QueryBook),#添加依赖#common:Depends(QueryBook)=Depends(),#添加依赖#common=Depends(QueryBook)#添加依赖):res=list()#使用page、limit组合返回值fornuminrange(common.page):res.append([{'book_name':fake_books_db[_]['book_name']}for_inrange(common.limit)])returnresif__name__=='__main__':uvicorn.run(app)
错误处理只需要使用raise关键字就可以返回错误响应触发HTTPException异常响应importuvicornfromfastapiimportFastAPIfromfastapi.exceptionsimportHTTPExceptionapp=FastAPI()@app.get('/http_exception')defexception(city:str):ifcity!='cd':raiseHTTPException(detail='cityisnotcd',#错误响应内容headers={'city_error':'cityisnotcd'},#发生错误添加的头部信息status_code=420)#自定义响应码return{'city':'cd'}if__name__=='__main__':uvicorn.run(app)重写错误响应准确一点是对相应的错误进行拦截修改，并不是重写重写HTTPException异常响应fromfastapi.exceptionsimportHTTPException#fromstarlette.exceptionsimportHTTPExceptionasStarletteHTTPException#与上面一排一样app=FastAPI()#在异常中间件拦截，相当于重写@app.exception_handler(HTTPException)asyncdefhttp_exception_v1(request:Request,exc:HTTPException):"""#改变成字符串响应:paramrequest:不可省略:paramexc:HTTPException:return:"""returnPlainTextResponse(str(exc.detail),status_code=400)重写RequestValidationError参数验证错误响应fromfastapi.exceptionsimportRequestValidationError#在异常中间件拦截RequestValidationError进行操作@app.exception_handler(RequestValidationError)asyncdefrequest_validation_error_v1(request:Request,exc:RequestValidationError):"""#改变成字符串响应:paramrequest:不可省略:paramexc:RequestValidationError:return:"""print(exc)returnPlainTextResponse(str(exc.errors()),status_code=500)自写错误处理类当提供的两种不够用的时候（HTTPException、RequestValidationError），可以自己写importuvicornfromfastapiimportFastAPIfromfastapi.requestsimportRequestfromstarlette.responsesimportPlainTextResponseapp=FastAPI()#自定义错误类继承Exception类classUnicornException(Exception):def__init__(self,error_city:str,status_code:int):self.error_city=error_cityself.status_code=status_code#异常中间件拦截自定义的异常@app.exception_handler(UnicornException)asyncdefrequest_validation_error_v1(request:Request,exc:UnicornException):"""#改变成字符串响应:paramrequest:不可省略:paramexc:UnicornException:return:"""print(exc)returnPlainTextResponse(str(exc.error_city),status_code=exc.status_code)@app.get('/validation')defvalidation(city:int):ifcity!=111#触发自定义异常raiseUnicornException(error_city='cityisnot111',#错误响应内容status_code=420)#自定义响应码return{'city':city}if__name__=='__main__':uvicorn.run(app)
应用常见配置操作主要对整个项目进行描述app=FastAPI(title='FastAPI标题',description='FastAPI学习教程分享',version='1.0.0',docs_url='/docs',#默认/docs可以自定义redoc_url='/redoc',#默认/redoc可以自定义)路径配置操作主要对接口进行描述fromfastapiimportFastAPIfrompydanticimportBaseModelimportuvicornfromstarletteimportstatusapp=FastAPI()classUser(BaseModel):user_name:strpassword:str@app.post('/path_operation_configuration',response_model=User,tags=['用户信息'],summary='接口概括',#description='接口的功能描述',#会覆盖下面接口函数注释的说明response_description='响应描述',#deprecated=True,#True表示弃用接口，文档显示灰色但还是可以用status_code=status.HTTP_200_OK)asyncdefpath(user:User):"""-接口的功能描述-支持MarkDown语法:paramuser:用户信息模型:return:响应用户信息"""returnuser.dict()if__name__=='__main__':uvicorn.run(app)接口弃用
使用静态文件常用于存放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)
说明Redis为订阅&发布模型如下，共有6个命令。相关命令subscribe订阅subscribechannel[channels]127.0.0.1:6379>subscribenews.itnews.sportReadingmessages...(pressCtrl-Ctoquit)1)"subscribe"2)"news.it"3)(integer)11)"subscribe"2)"news.sport"3)(integer)2#订阅后客户端挂起等待接受消息psubscribe模式订阅psubscribe与subscribe的差别只有一个地方，其他都一样subscribe：只能订阅指定名称的频道psubscribe：可以使用通配符指定频道#客户端1使用模式订阅127.0.0.1:6379>psubscribenews.*Readingmessages...(pressCtrl-Ctoquit)1)"psubscribe"2)"news.*"3)(integer)1#接收到的消息1)"pmessage"2)"news.*"3)"news.a"4)"testa"1)"pmessage"2)"news.*"3)"news.b"4)"testb"#客户端2发送消息127.0.0.1:6379>publishnews.atesta(integer)1127.0.0.1:6379>publishnews.btestb(integer)1publish发送消息publishchannelmessage发送消息到指定的频道127.0.0.1:6379>publishnews.ittest(integer)1#发送成功127.0.0.1:6379>publishxxtest(integer)0#发送失败pubsub查看订阅状态pubsub<子命令>pubsubchannels：默认查看所有使用subscribe命令订阅的频道，可以使用匹配127.0.0.1:6379>pubsubchannels1)"news.sport"2)"news.it"127.0.0.1:6379>pubsubchannelsnews.*1)"news.sport"2)"news.it"pubsubnumsubchannel[channel]：查看指定频道订阅的数量127.0.0.1:6379>pubsubnumsubnews.itnews.sport1)"news.it"2)(integer)13)"news.sport"4)(integer)1pubsubnumpat：返回使用psubscribe命令客户端的总数量#客户端1订阅127.0.0.1:6379>psubscribenews.goodReadingmessages...(pressCtrl-Ctoquit)1)"psubscribe"2)"news.good"3)(integer)1#客户端2查看psubscribe订阅的数量127.0.0.1:6379>pubsubnumpat(integer)1127.0.0.1:6379>unsubscribe退订指定的频道unsubscribechannel[channels]127.0.0.1:6379>unsubscribenews.it#退订news.it频道1)"unsubscribe"#退订2)"news.it"#退订的频道名称3)(integer)0#订阅的数量punsubscribe退订使用模式订阅频道punsubscribechannel[channels]的差别只有一个地方，其他都一样unsubscribe：只能退订指定名称的频道punsubscribe：可以使用通配符退订频道127.0.0.1:6379>punsubscribenews.*1)"punsubscribe"#退订2)"news.*"#退订的频道名称3)(integer)0#订阅的数量