FastAPI子依赖传参返回及鉴权后权限校验方案咨询

首先明确一个FastAPI依赖系统的核心机制，也是你当前顾虑的核心解答：

同一次HTTP请求的生命周期内，FastAPI会对所有声明的相同依赖做结果缓存。只要依赖指向同一个可调用对象，无论你在路径函数、多少层嵌套子依赖中重复声明，整个请求周期内依赖逻辑只会执行一次，所有位置拿到的都是同一个返回实例。你担心的重复创建连接、嵌套事务问题，在依赖指向同一个可调用对象的前提下完全不会发生。

具体实现步骤 #

1. 统一数据库连接依赖入口 #

首先废弃你当前同时存在的engine_begin、engine_connect两个独立的连接依赖，只保留一个全局唯一的数据库连接依赖，统一管理连接的获取、事务提交/回滚、连接释放逻辑：

from sqlalchemy.ext.asyncio import AsyncConnection
from fastapi import Depends

async def get_db() -> AsyncConnection:
    # 统一在依赖内管理事务生命周期
    async with engine.begin() as conn:
        yield conn

后续所有需要数据库连接的位置，统一使用conn: AsyncConnection = Depends(get_db)声明即可。

2. 按单一职责拆分各层依赖 #

严格约束每个依赖的返回值范围，只返回该依赖职责内生成的对象，禁止跨职责传递资源：

• JWT认证依赖get_current_user：仅负责解析校验JWT令牌，不访问数据库，只返回令牌中携带的基础用户标识信息

  from jose import JWTError, jwt
  from fastapi import HTTPException, status
  
  async def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
      try:
          payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
          user_id = payload.get("sub")
          if not user_id:
              raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
      except JWTError:
          raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
      return {"user_id": user_id, "username": payload.get("username")}
  

• 超级管理员权限校验依赖get_superuser：仅负责权限校验，依赖get_current_user获取基础用户信息、依赖get_db获取数据库连接查询用户权限，校验通过后仅返回完整的超级用户实体，不返回数据库连接

  async def get_superuser(
      current_user: dict = Depends(get_current_user),
      conn: AsyncConnection = Depends(get_db)
  ) -> dict:
      query = select(user_table).where(user_table.c.id == current_user["user_id"])
      result = await conn.execute(query)
      user = result.scalar_one_or_none()
      if not user or not user.is_superuser:
          raise HTTPException(
              status_code=status.HTTP_403_FORBIDDEN,
              detail="Insufficient permissions"
          )
      return dict(user._mapping)
  

3. 路径函数按需声明依赖 #

业务接口中可以根据需要，同时声明超级用户依赖和数据库连接依赖，不需要做任何特殊处理：

@app.post("/admin/do-something")
async def do_something(
    superuser: dict = Depends(get_superuser),
    conn: AsyncConnection = Depends(get_db)
):
    # 此处拿到的conn与get_superuser内部校验时使用的是同一个连接实例，归属同一个外层事务
    # 直接执行业务SQL即可
    await conn.execute(...)
    return {"code": 0, "msg": "success"}

方案优势 #

• 职责完全分离：每个依赖只负责单一逻辑，认证、授权、连接管理三者完全解耦，不存在从用户对象中提取数据库连接这类反直觉设计
• 无额外性能开销：依赖缓存机制保证整个请求周期只创建一个数据库连接、开启一个事务，不会出现资源浪费或嵌套事务问题
• 复用性强：普通用户接口只需声明Depends(get_current_user)，管理员接口声明Depends(get_superuser)，任意需要操作数据库的位置都可以直接声明Depends(get_db)拿到当前请求的连接，不需要做额外的参数透传
• 链路清晰：依赖调用链完全透明，顺着声明就能看到每一层的依赖关系，没有隐式的资源传递，后续维护成本极低

需要避开的反模式 #

• 不要在权限、认证类依赖中返回数据库连接这类通用资源，会导致依赖约定隐式化，后续维护者很容易因为不知道这个隐式约定重复创建连接
• 不要在路径装饰器的dependencies参数中声明数据库连接依赖：装饰器级别的依赖只会执行逻辑，不会将返回值纳入依赖缓存供其他子依赖、路径函数获取
• 不要创建多个功能重叠的数据库连接依赖：不同的可调用对象会被FastAPI识别为不同的依赖，分别执行逻辑，真的会产生多个连接和嵌套事务。

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