如何在 FastAPI 中统一处理用户认证与数据库会话，并实现可测试的全局依赖-Python教程-PHP中文网

如何在 FastAPI 中统一处理用户认证与数据库会话，并实现可测试的全局依赖

本文介绍如何通过 FastAPI 全局依赖替代手动中间件，安全高效地为每个请求注入数据库会话和当前用户对象，并提供基于 async_sessionmaker 和 dependency_overrides 的可回滚单元测试方案。

本文介绍如何通过 fastapi 全局依赖替代手动中间件，安全高效地为每个请求注入数据库会话和当前用户对象，并提供基于 `async_sessionmaker` 和 `dependency_overrides` 的可回滚单元测试方案。

在 FastAPI 应用中，为每个请求自动加载数据库会话并获取当前用户，不应依赖自定义 ASGI 中间件（如 BaseHTTPMiddleware）——这不仅违背 FastAPI 的依赖注入设计哲学，更会导致测试困难、会话生命周期失控及类型提示丢失等问题。推荐采用 全局依赖（Global Dependencies） + 依赖覆盖（Dependency Overrides） 的标准实践，兼顾生产健壮性与测试隔离性。

✅ 正确做法：用全局依赖替代中间件

首先，移除 SessionMiddleware 和 AuthenticationMiddleware，改用 FastAPI 原生依赖机制：

from fastapi import Depends, HTTPException, Request, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated

# 1. 定义会话依赖（通常来自 async_sessionmaker）
async def get_db() -> AsyncSession:
    # 实际应从 async_sessionmaker() 获取，此处为示意
    async with AsyncSessionLocal() as session:
        yield session

# 2. 定义用户依赖（自动从会话中查询）
async def get_current_user(
    request: Request,
    db: AsyncSession = Depends(get_db)
) -> User:
    client_name = request.headers.get("X-User-Name")  # 或从 JWT/cookie 解析
    if not client_name:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing user identity")

    user = await db.execute(select(User).where(User.name == client_name))
    user = user.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return user

# 3. 注册为全局依赖（所有路由默认应用）
app = FastAPI(
    dependencies=[Depends(get_db), Depends(get_current_user)]
)

该方式天然支持类型提示、异步上下文管理、自动异常传播，且与 OpenAPI 文档、依赖注入链完全兼容。

? 可测试性关键：复用同一会话实例

测试时需确保 测试函数 与 被测端点内部使用的 AsyncSession 是同一个实例，从而支持事务级回滚。核心在于：

使用 @event.listens_for(..., "after_transaction_end") 监听嵌套事务结束，自动开启新保存点；
通过 app.dependency_overrides 替换原始 get_db，注入测试专用会话工厂；
所有操作在单个连接内完成，rollback() 即可彻底清理。

以下为精简可靠的测试 fixture 示例：

import pytest
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy import text
from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncConnection
from fastapi.testclient import TestClient
from httpx import AsyncClient

@pytest.fixture(scope="function", autouse=True)
async def override_get_db(session: AsyncSession):
    app.dependency_overrides[get_db] = lambda: session
    yield
    app.dependency_overrides.clear()

@pytest.fixture
async def session() -> AsyncGenerator[AsyncSession, None]:
    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
        AsyncSessionLocal = async_sessionmaker(
            bind=conn,
            expire_on_commit=False,
            class_=AsyncSession,
        )
        async_session = AsyncSessionLocal()

        # 关键：启用嵌套事务（savepoint）
        @event.listens_for(async_session.sync_session, "after_transaction_end")
        def reopen_nested_transaction(session: Session, transaction):
            if conn.in_nested_transaction():
                conn.sync_connection.begin_nested()

        yield async_session

        await async_session.close()
        await conn.rollback()
        await conn.close()
    await engine.dispose()

⚠️ 注意事项：

勿在中间件中手动管理 session.close()：FastAPI 依赖生命周期自动处理资源释放；

避免 request.state 存储会话或用户：破坏依赖注入透明性，导致类型检查失效、IDE 无法推导；

测试中必须使用 AsyncClient（非 TestClient）：同步客户端无法正确执行 async def 依赖；

全局依赖需显式声明 autouse=True：否则仅对显式 Depends() 的路由生效。