FastAPI Python中间件中如何获取当前已认证的活跃用户

核心问题原因 #

FastAPI 的 Depends 依赖解析仅在路由匹配成功后、路径操作函数执行前的阶段运行，HTTP 中间件的运行逻辑不接入 FastAPI 的依赖注入体系，因此你尝试的两种方案都无法拿到实际的用户对象。

可行解决方案 #

方案1：中间件内手动解析token（无需修改现有认证逻辑） #

你可以直接在中间件中复用JWT解析逻辑，手动从请求头提取token完成用户获取，不需要依赖Depends体系：

@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
    
    # 手动解析token获取用户
    user = "匿名用户"
    authorization = request.headers.get("Authorization")
    if authorization and authorization.startswith("Bearer "):
        token = authorization.split(" ")[1]
        try:
            payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
            username = payload.get("sub")
            if username:
                user_obj = UserManager.get_user(real_user_db, username=username)
                if user_obj and not user_obj.disabled:
                    user = username
        except JWTError:
            # token无效时保持匿名，后续路由逻辑会处理认证报错
            pass

    db.write_log(path, method, params, host, process_time, user)
    response.headers["X-Process-Time"] = str(process_time)
    return response

该方案不需要修改现有路由和认证逻辑，适配公开接口和需要认证的接口，没有token的请求会默认记录为匿名用户。

方案2：通过Request.state传递用户（逻辑复用性更高） #

如果不想重复写JWT解析逻辑，可以通过请求对象的state属性传递用户信息：

1. 先定义一个全局依赖，将当前用户写入请求state

async def inject_current_user(request: Request, token: str = Depends(oauth2_scheme)):
    # 没有token的公开接口可以捕获异常，写入匿名用户
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username:
            user_obj = UserManager.get_user(real_user_db, username=username)
            if user_obj and not user_obj.disabled:
                request.state.current_user = username
                return
    except (JWTError, HTTPException):
        pass
    request.state.current_user = "匿名用户"

2. 将该依赖注册为全局依赖，所有路由执行前都会自动运行：

app = FastAPI(dependencies=[Depends(inject_current_user)])

3. 中间件中直接从request.state读取用户即可：

@app.middleware("http")
async def log_things(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    # 其他信息获取逻辑不变
    user = request.state.current_user
    db.write_log(path, method, params, host, process_time, user)
    response.headers["X-Process-Time"] = str(process_time)
    return response

该方案的优势是认证逻辑只需要维护一份，后续如果认证规则修改，只需要调整inject_current_user即可。

内容的提问来源于stack exchange，提问作者user10754243