python - 如何在中间件FastAPI python中获取当前活跃用户

Question

我已经使用此处的文档在 FastAPI Python 上开发了一个身份验证 https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/ (我完全使用显示的代码，没什么特别的，auth 工作起来很有魅力)；

现在，我需要在每次调用每条路由时在数据库中记录一些事情:请求花费的时间、当前请求的方法等 - 以及用户名(使用该路径的用户名)。

为此，我开发了一个如下所示的 http 中间件:

@app.middleware("http")
async def log_things(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)

    path = request.url.path
    method = request["method"]
    params = request.path_params
    host = request.client.host + ":" + str(request.client.port)
    process_time = time.time() - start_time
    user = await security_management.get_current_active_user()  
    user = "dev" # MOCK todo how to get user

    db.write_log(path, method, params, host, process_time, user)

    response.headers["X-Process-Time"] = str(process_time)
    return response

问题:我无法检索用户名。我检索的内容封装在一个我无法解析的“Depend”对象中。

我尝试了一些事情:

• 在 Depend 上使用 await，
• 使用 current_user: User = Depends(get_current_user) 作为我的中间件函数的参数，

我有点卡住了，无法从这个 Depend 对象中提取有值(value)的数据，即使它在安全部分用下面的这两个函数工作得很好(我也试图在参数方面重现......)

async def get_current_user(
        token: str = Depends(oauth2_scheme),
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = UserManager.get_user(real_user_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user # I can read current_user well...

有什么想法吗？谢谢:)

Answer 1

我在我的应用程序中遇到了完全相同的问题，并找到了解决方法/解决方案。

    @app.middleware("http")
    async def request_middleware(request, call_next):
    
        # some operation
    
        if request.headers.get('Authorization'):
            HttpRequestUtil.set_current_user_context(request=request)
    
        return await call_next(request)


    class HttpRequestUtil:
    
        @staticmethod
        def get_bearer_token(request: Request):
            auth_token = request.headers.get('Authorization')
            if 'Bearer' in auth_token:
                bearer_token: str = auth_token.split('Bearer')[1].strip()
                return bearer_token
    
        @staticmethod
        def set_current_user_context(request: Request):
            jwt_token=HttpRequestUtil.get_bearer_token(request)
            
            # if you want jwt claims, you can do below operation
            jwt_claims = jwt.get_unverified_claims(jwt_token)
            # ...
            # ...
            # Other Operation
            # ...