You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在FastAPI路由中声明依赖并注入当前用户到处理函数?

在FastAPI路由层统一配置Bearer Token认证并注入当前用户

你完全可以在路由层统一配置认证依赖,同时让处理函数轻松获取当前用户,不用在每个函数里重复声明Security(get_current_active_user),下面提供两种实用方案:

方案一:利用request.state存储用户

通过修改认证依赖,将用户信息存入请求对象的state属性中,后续处理函数直接从request.state读取即可。

步骤1:修改认证依赖函数

调整get_current_active_user,让它接收Request对象,并将验证后的用户存入request.state:

from fastapi import Request, Depends, HTTPException, status
from jose import JWTError, jwt
# 替换成你的实际密钥和算法
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

async def get_user(username: str):
    # 这里替换为你的用户查询逻辑,比如从数据库获取
    return {"username": username, "is_active": True}

async def get_current_active_user(request: Request):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="凭证验证失败",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # 提取Bearer Token
        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            raise credentials_exception
        token = auth_header.split(" ")[1]
        # 解析Token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        # 获取用户并验证是否激活
        user = await get_user(username=username)
        if not user.get("is_active", False):
            raise credentials_exception
        # 将用户存入request.state
        request.state.current_user = user
        return user
    except JWTError:
        raise credentials_exception

步骤2:路由层配置统一依赖

保持你原来的路由配置即可,认证依赖会在每个请求处理前执行:

app.include_router(
    users_router,
    tags=["Users"],
    prefix="",
    dependencies=[Depends(get_current_active_user)],
)

步骤3:处理函数中获取用户

在处理函数里注入Request对象,直接从request.state读取当前用户:

from fastapi import Request

@users_router.get("/me")
async def read_users_me(request: Request):
    return request.state.current_user

方案二:创建专用的用户获取依赖

如果觉得直接操作Request不够直观,可以封装一个小依赖,专门从request.state提取用户,处理函数直接注入该依赖即可:

步骤1:新增用户获取依赖

from fastapi import Request

async def get_current_user(request: Request):
    return request.state.current_user

步骤2:处理函数中使用该依赖

@users_router.get("/me")
async def read_users_me(current_user = Depends(get_current_user)):
    return current_user

这种方式更贴近你原来的函数写法,但避免了重复声明认证逻辑。

注意事项

  • 路由层的dependencies会对该路由下的所有路径操作生效,确保每个请求都会先执行认证逻辑,存入用户信息。
  • 如果你的认证依赖本身返回用户对象,路由层的依赖会执行它,但不会自动将结果传递给处理函数,所以必须通过request.state或者类似方式存储,才能让处理函数获取到。

内容的提问来源于stack exchange,提问作者p.magalhaes

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.03 09:00:23