什么是OAuth2认证规范：共同准守的规范，需要学习这种规范、流程OAuth2是一个规范，它定义了几种处理身份认证和授权的方法。基本的规范结构不标准的代码，只是讲解规范需要校验token的接口，把OAuth2PasswordBearer实例化对象作为依赖就好。fromfastapi.securityimportOAuth2PasswordBearer,OAuth2PasswordRequestForm"""1、OAuth2PasswordBearer只是接收URL作为参数的一个类，\客户端会向该URL发送用于请求token的参数（username、password...）2、OAuth2PasswordBearer不会创建相应的URL路径操作，\只是指明客户端用来请求Token的URL地址3、把OAuth2PasswordBearer的实例作为依赖，\FastAPI会检查请求的Authorization头信息，如果没有找到会返回401状态码（UNAUTHORIZED）"""#实例化OAuth2PasswordBearer指明请求Token地址oauth2_scheme=OAuth2PasswordBearer(tokenUrl='token')#http://127.0.0.1:8000/token#请求token时具体的处理@app.post('/token')#对应上面的OAuth2PasswordBearer实例化参数asyncdefget_token(form_data:OAuth2PasswordRequestForm=Depends()):"""这里使用了OAuth2规范的表单类""""""OAuth2规范的Token返回结构"""return{'access_token':user.username,'token_type':'bearer'}#使用oauth2_scheme作为依赖，会自动校验请求的Authorization头信息asyncdefget_user_info(user_token:str=Depends(oauth2_scheme)):returnuser_token#使用get_user_info作为依赖，直接请求会返回401状态码（UNAUTHORIZED）@app.post('/user/me')asyncdefread_user_me(user:str=Depends(get_user_info)):returnuser实例项目中不会这样使用（便于理解），项目更多的是使用JsonWebToken认证模式importuvicornfrompydanticimportBaseModelfromfastapiimportDepends,FastAPI,HTTPException,statusfromfastapi.securityimportOAuth2PasswordBearer,OAuth2PasswordRequestFormapp=FastAPI()fake_user_db={#模拟用户数据库'Bob':{'username':'Bob','password':'123456'},'Mary':{'username':'Mary','password':'987654'}}classUser(BaseModel):username:strpassword:str#实例化OAuth2PasswordBearer类，指明请求token接口地址oauth2_scheme=OAuth2PasswordBearer(tokenUrl='token')@app.post('/token')#获取token的接口具体方法asyncdefget_token(form_data:OAuth2PasswordRequestForm=Depends()):user_dict=fake_user_db.get(form_data.username)#数据库获取用户ifnotuser_dict:#检查用户是否存在raiseHTTPException(status_code=status.HTTP_400_BAD_REQUEST,detail='Incorrectusernameorpassword')user=User(**user_dict)ifform_data.password!=user.password:#检查密码是否正确raiseHTTPException(status_code=status.HTTP_400_BAD_REQUEST,detail='Incorrectusernameorpassword')"""access_token是返回的token（这里是为了方便，所以使用username作用token）token_type是对应实例化OAuth2Password'Bearer'"""return{'access_token':user.username,'token_type':'bearer'}#依赖oauth2_scheme自动校验Authorization头信息asyncdefget_user_info(user_token:str=Depends(oauth2_scheme)):returnUser(**fake_user_db[user_token])#获取token的用户详细信息@app.post('/user/me')asyncdefread_user_me(user:User=Depends(get_user_info)):returnuserif__name__=='__main__':uvicorn.run(app)接口测试页面直接请求，验证失败获取Token获取Token用户信息直接请求获取Token接口（@app.post(‘/token’)）
说明有时候我们并不需要依赖注入的返回值，这时我们就可以从路径操作注入依赖路径操作装饰器依赖注入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)