FastAPI Request中存储信息

发布时间 2023-09-22 14:32:28作者: snake_j

在请求前拦截,检测token

使用 Request.state.XXX 存信息

from fastapi import Request, FastAPI              
from py_jwt.use_jwt import decode_token           
                                                  
                                                  
def register_middleware(app: FastAPI):                                                              
    @app.middleware("http")                       
    async def dispath(r: Request, call_next):     
        '''                                       
        地址是 /user/get_token 直接通过           
        其余需要检测token                         
        '''                                                                                         
        if r.url.path == "/user/get_token":       
            response = await call_next(r)         
            return response                       
        else:                                     
            if "Authorization" in r.headers:      
                jwt = r.headers["authorization"]  
                get_jwt = jwt.split("jwt ")[-1]   
                ret = decode_token(get_jwt)       
                                                  
                if ret["ret"] == 1:               
                    # 信息添加到 Request.state    
                    r.state.isok = 1              
                    r.state.token = ret["result"] 
                    response = await call_next(r) 
                    return response               
                else:                             
                    r.state.isok = 0              
                    r.state.msg = ret["result"]   
                    response = await call_next(r) 
                    return response               
            else:                                 
                r.state.isok = 0                  
                r.state.msg = "no token"          
                response = await call_next(r)     
                return response                   

函数中读取

@router.get("/home")
def user_home(r: Request):
    # 读取信息
    if r.state.isok == 0:
        return {"result": r.state.msg}

    print(r.state.token)
    return {"result": "user home"}