在FastAPI中实现带JWT认证的异步端点并管理数据库会话

题目

在FastAPI中实现带JWT认证的异步端点并管理数据库会话

信息

• 类型：问答
• 难度：⭐⭐

考点

依赖注入,异步编程,JWT认证,数据库会话管理

快速回答

实现步骤：

1. 使用Depends创建数据库会话依赖
2. 通过OAuth2PasswordBearer处理JWT令牌
3. 编写认证依赖项验证JWT并获取当前用户
4. 在异步路由中使用async def和上述依赖
5. 使用SQLAlchemy Core异步查询数据库

关键代码：

# 依赖项示例
get_db = ...
get_current_user = ...

@app.get("/items/")
async def read_items(db: Session = Depends(get_db), 
                   user: User = Depends(get_current_user)):
    items = await db.execute(select(Item))
    return items.scalars().all()

1. 核心原理说明

FastAPI的依赖注入系统允许声明路由操作所需的组件（如数据库会话或认证对象）。结合异步编程时：

• JWT认证流程：客户端在Authorization头携带Bearer令牌 → FastAPI通过依赖项验证令牌有效性 → 解码获取用户数据
• 数据库会话管理：使用async依赖项创建每个请求独立的会话，确保线程安全
• 异步优势：I/O操作（如数据库查询）使用await释放事件循环，提升并发能力

2. 完整代码实现

依赖项设置

# 数据库会话依赖 (使用SQLAlchemy 1.4+异步模式)
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

# JWT认证依赖
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme), 
                          db: Session = Depends(get_db)) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid credentials")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

异步端点定义

@app.get("/protected-data", response_model=List[Item])
async def get_protected_data(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    # 异步数据库查询
    result = await db.execute(select(Item).where(Item.owner_id == current_user.id))
    return result.scalars().all()

3. 最佳实践

• 会话生命周期：使用yield依赖确保会话正确关闭（即使发生异常）
• 密钥管理：SECRET_KEY应从环境变量加载，切勿硬编码
• 异步兼容性：数据库驱动需支持异步（如asyncpg对应PostgreSQL）
• 错误处理：为HTTPException添加详细错误信息，但避免泄露敏感数据

4. 常见错误

• 阻塞操作：在异步端点中调用同步I/O（如未适配的同步数据库驱动）
• 会话泄露：忘记yield或async with导致会话未关闭
• JWT配置错误：服务端与客户端算法不一致或密钥不匹配
• 依赖顺序：get_current_user依赖get_db时需注意声明顺序

5. 扩展知识

• 性能优化：使用redis缓存JWT黑名单或高频查询数据
• 安全增强：
  • JWT添加过期时间(exp)和发行者(iss)校验
  • 敏感路由强制HTTPS
• 进阶方案：
  • 使用fastapi-jwt-auth等第三方库简化JWT流程
  • 基于Middleware实现全链路日志追踪