Python极客

什么是中间件中间件是一个函数请求达到响应逻辑之前会经过一层或多层中间件,响应结果返回客户端也会经过一层或多层中间件FastAPI中间件开发自定义中间件importtimeimportuvicornfromfastapiimportFastAPIfromfastapi.requestsimportRequestfromfastapi.responsesimportResponseapp=FastAPI()@app.middleware('http')#中间件类型现在只支持http类型asyncdefadd_process_time_header(request:Request,call_next):"""计算请求响应时间:paramrequest:请求:paramcall_next:请求处理回调:return:"""start_time=time.time()res:Response=awaitcall_next(request)process_time=time.time()-start_timeres.headers['X-Process-Time']=str(process_time)returnres@app.get('/middleware')defmiddleware():return'没有处理逻辑,响应成功'if__name__=='__main__':uvicorn.run(app)提供的中间件CORSMiddleware跨域资源共享中间件fromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewareapp=FastAPI()app.add_middleware(CORSMiddleware,allow_origins=[#允许访问的域域-->协议:域名:端口'http://127.0.0.1','http://127.0.0.1:8000',],allow_methods=['*','GET'],#*可以通配allow_headers=['*'],#头部allow_credentials=False,#HTTPS证书allow_origin_regex=None,#正则表达式匹配'https://.*\.example\.orgexpose_headers=[],#指明可以被浏览器访问的响应头max_age=600#设定浏览器缓存CORS响应的最长时间,单位是秒。默认为600)@app.get("/")asyncdefmain():return{"message":"HelloWorld"}HTTPSRedirectMiddleware强制所有传入请求必须是https或wss如果不是https则会自动跳转到https,如果网站没有配置https,则报错h11._util.RemoteProtocolError:illegalrequestlineWARNING:InvalidHTTPrequestreceived.fromfastapiimportFastAPIfromfastapi.middleware.httpsredirectimportHTTPSRedirectMiddlewareapp=FastAPI()app.add_middleware(HTTPSRedirectMiddleware)@app.get("/")asyncdefmain():return{"message":"HelloWorld"}TrustedHostMiddleware强制所有传入请求都具有正确设置的Host标头,以防止HTTP主机标头攻击。如果不包含,则响应400状态码,并返回InvalidhostheaderfromfastapiimportFastAPIfromfastapi.middleware.trustedhostimportTrustedHostMiddlewareapp=FastAPI()app.add_middleware(TrustedHostMiddleware,allowed_hosts=["bigdataboy.cn","*.bigdataboy.cn"]#允许的hosts列表)@app.get("/")asyncdefmain():return{"message":"HelloWorld"}GZipMiddleware处理包含”gzip”在Accept-Encoding标头中的任何请求的GZip响应。fromfastapiimportFastAPIfromfastapi.middleware.gzipimportGZipMiddlewareapp=FastAPI()app.add_middleware(GZipMiddleware,minimum_size=1000)#不要GZip响应小于此最小字节大小。默认为500.@app.get("/")asyncdefmain():return{"message":"HelloWorld"}带yield关键字依赖,依赖中的退出代码将在执行中间件后执行defget_db():db=dbSession()try:yielddbfinally:db.close()

Python极客

什么是JsonWebTokens(JWT)请求流程首先浏览器向服务器发送请求得到token然后浏览器每次请求都自动在头部带上tokentoken的组成token是一些信息(用户名,过期时间…)的加密结果,加密的秘钥保存在服务器,因此只要秘钥不被泄漏,就认为是安全的。FastAPI的JWT哈希密码把用户的明文密码,加密保存在数据库,即使密码泄漏,也不知道用户真正的密码。推荐的算法是「Bcrypt」:pip3installpasslib[bcrypt]frompasslib.contextimportCryptContext#创建哈希上下文pwd_context=CryptContext(schemes=["bcrypt"],deprecated="auto")#校验密文明文defverify_password(plain_password,hashed_password):returnpwd_context.verify(plain_password,hashed_password)#加密明文defget_password_hash(password):returnpwd_context.hash(password)if__name__=='__main__':password_hash=get_password_hash('123456')#加密结果是动态的每次不一样,但是验证是一样的print(password_hash)print(verify_password('123456',password_hash))token生成&校验逻辑说明需要加密使用的秘钥加密算法token过期时间#opensslrand-hex32SECRET_KEY="12b2e365a1b19051f115e46e8dfd7200e63510319a791bcb2dcf605626e1aa0c"ALGORITHM="HS256"ACCESS_TOKEN_EXPIRE_MINUTES=30token生成与校验安装模块:pipinstallpython-jose[cryptography]fromtypingimportOptionalfromjoseimportJWTError,jwtfromdatetimeimportdatetime,timedelta#opensslrand-hex32SECRET_KEY="12b2e365a1b19051f115e46e8dfd7200e63510319a791bcb2dcf605626e1aa0c"ALGORITHM="HS256"ACCESS_TOKEN_EXPIRE_MINUTES=30defcreate_access_token(user:dict,expires_delta:Optional[timedelta]=None):to_encode=user.copy()#浅拷贝:深拷贝父对象(一级目录),子对象(二级目录)不拷贝,还是引用ifexpires_delta:#能通过参数指定过期时间expire=datetime.utcnow()+expires_deltaelse:expire=datetime.utcnow()+timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)to_encode.update({"exp":expire})#加入过期时间encoded_jwt=jwt.encode(to_encode,SECRET_KEY,algorithm=ALGORITHM)#加密信息得到tokenreturnencoded_jwtif__name__=='__main__':access_token_expires=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)#生成时间格式0:30:00token=create_access_token(#生成Token{'username':'Bob'},expires_delta=access_token_expires)print(f'user-->token:{token}')to_decode=jwt.decode(token,SECRET_KEY,algorithms=[ALGORITHM])#解密tokenprint(f'token-->user{to_decode}')案例此案例结合上篇的基于PasswordBearerToken的OAuth2认证获取TokenimportuvicornfromtypingimportOptionalfrompydanticimportBaseModelfromdatetimeimportdatetime,timedeltafromfastapiimportDepends,FastAPI,HTTPException,statusfromfastapi.securityimportOAuth2PasswordBearer,OAuth2PasswordRequestFormfromjoseimportJWTError,jwt#token使用frompasslib.contextimportCryptContext#哈希密码#opensslrand-hex32SECRET_KEY="12b2e365a1b19051f115e46e8dfd7200e63510319a791bcb2dcf605626e1aa0c"ALGORITHM="HS256"ACCESS_TOKEN_EXPIRE_MINUTES=30fake_user_db={#模拟用户数据库'Bob':{'username':'Bob','hash_password':'$2b$12$Qm6i.pPxCM/Kc672T0GJZOr6Wnq2YkjZm.UMk1O9abq.8fx3fas52'},#明文123456'Mary':{'username':'Mary','hash_password':'$2b$12$Qm6i.pPxCM/Kc672T0GJZOr6Wnq2YkjZm.UMk1O9abq.8fx3fas52'},#明文123456}classUser(BaseModel):username:strhash_password:strclassToken(BaseModel):access_token:strtoken_type:str='bearer'pwd_context=CryptContext(schemes=["bcrypt"],deprecated="auto")#实例化OAuth2PasswordBearer类,指明请求token接口地址oauth2_scheme=OAuth2PasswordBearer(tokenUrl='/jwt/token')app=FastAPI()#验证密码defverify_password(plain_password:str,hash_password)->bool:returnpwd_context.verify(plain_password,hash_password)defget_password_hash(password:str):returnpwd_context.hash(password)defget_user(db,username:str)->User:ifusernameindb:user_dict=db[username]returnUser(**user_dict)defauthenticate_user(fake_db,username:str,password:str):user=get_user(fake_db,username)ifnotuser:returnFalseifnotverify_password(password,user.hash_password):returnFalsereturnuserdefcreate_access_token(user:dict,expires_delta:Optional[timedelta]=None):to_encode=user.copy()#浅拷贝:深拷贝父对象(一级目录),子对象(二级目录)不拷贝,还是引用ifexpires_delta:#能通过参数指定过期时间expire=datetime.utcnow()+expires_deltaelse:expire=datetime.utcnow()+timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)to_encode.update({"exp":expire})#加入过期时间encoded_jwt=jwt.encode(to_encode,SECRET_KEY,algorithm=ALGORITHM)#加密信息得到tokenreturnencoded_jwt@app.post('/jwt/token')#获取token的接口具体方法asyncdefget_token(form_data:OAuth2PasswordRequestForm=Depends()):user=authenticate_user(#获取用户对象fake_db=fake_user_db,username=form_data.username,password=form_data.password)ifnotuser:raiseHTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail='Incorrectusernameorpassword',headers={'WWW-Authenticate':'Bearer'})#规范access_token_expires=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)access_token=create_access_token(#生成Tokenuser=user.dict(),expires_delta=access_token_expires)returnToken(access_token=access_token,token_type='bearer')if__name__=='__main__':uvicorn.run(app)获取当前用户信息asyncdefget_current_user(token:str=Depends(oauth2_scheme)):credentials_exception=HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Couldnotvalidatecredentials",headers={"WWW-Authenticate":"Bearer"},)try:#解密token获取用户payload=jwt.decode(token,SECRET_KEY,algorithms=[ALGORITHM])username:str=payload.get('username')ifusernameisNone:raisecredentials_exceptiontoken_data=usernameexceptJWTError:returncredentials_exceptionuser=get_user(fake_user_db,token_data)ifuserisNone:raisecredentials_exceptionreturnuser@app.post('/jwt/me')asyncdefread_users_me(current_user:User=Depends(get_current_user)):returncurrent_user获取Token获取当前用户信息

Python极客

什么是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’))

Python极客

什么是依赖注入依赖注入常用于以下场景共享业务逻辑(复用相同的代码逻辑)共享数据库连接实现安全、验证、角色权限等……上述场景均可以使用依赖注入,将代码重复最小化。依赖注入的作用复用相同代码、简化代码初步使用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)

Python极客

错误处理只需要使用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)