本文介绍如何通过 FastAPI 全局依赖替代中间件方式实现请求级用户加载与数据库会话管理，并提供基于 pytest + SQLAlchemy 异步会话的端到端可回滚测试方案，确保测试隔离性与生产逻辑一致性。
本文介绍如何通过 fastapi 全局依赖替代中间件方式实现请求级用户加载与数据库会话管理，并提供基于 `pytest` + sqlalchemy 异步会话的端到端可回滚测试方案，确保测试隔离性与生产逻辑一致性。
在现代 FastAPI 应用中，为每个请求自动加载当前用户并绑定数据库会话，不应依赖自定义中间件（尤其是涉及状态传递与生命周期管理时），而应优先采用 全局依赖（Global Dependencies） —— 它更符合 FastAPI 的设计哲学，具备类型安全、依赖注入链清晰、易于测试和调试等优势。
✅ 推荐架构：全局依赖替代中间件
首先，移除 SessionMiddleware 和 AuthenticationMiddleware。取而代之的是两个分层依赖：
-
get_db()：生成并管理异步数据库会话（AsyncSession），由 FastAPI 依赖注入系统自动生命周期管理； -
get_current_user()：接收db: AsyncSession，从请求头（如Authorization: Bearer <token></token>）解析凭证，查询数据库获取用户对象，并挂载至request.state.user或直接返回。
from fastapi import Depends, Request, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from api.database.setup import get_session # your async session factory
from api.models import User
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with get_session() as session:
yield session
async def get_current_user(
request: Request,
db: AsyncSession = Depends(get_db)
) -> User:
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing or invalid token")
token = auth_header[7:]
# 示例：根据 token 查询用户（实际应结合 JWT 解析或 OAuth2 验证）
user = await db.execute(select(User).where(User.token == token))
user = user.scalar_one_or_none()
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
request.state.user = user # 可选：供后续中间件/路由访问
return user
# 全局注册（main.py 中）
app = FastAPI(
dependencies=[Depends(get_current_user)] # ← 所有路由默认执行用户认证！
)
该方案天然支持依赖覆盖（app.dependency_overrides），无需修改任何路由签名，也避免了中间件中手动操作 request.state 带来的类型不安全与调试困难。
? 测试：共享会话 + 自动回滚的可靠方案
测试的核心挑战在于：让测试函数与被测路由使用同一个数据库会话实例，并在测试结束时原子性回滚所有变更。关键在于利用 SQLAlchemy 的嵌套事务（BEGIN SAVEPOINT）配合 @event.listens_for 动态恢复保存点。
以下是一个精简可靠的 session fixture 示例（适配异步场景）：
import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.event import listens_for
from sqlalchemy.engine import Engine
from api.main import app
from api.database.setup import get_session
@pytest.fixture
async def session() -> AsyncSession:
# 使用测试专用数据库 URL（如 test.db 或 test DB 实例）
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False, connect_args={"check_same_thread": False})
# 创建异步会话工厂
AsyncSessionLocal = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
# 启动连接并开启嵌套事务
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # 初始化表
async with AsyncSessionLocal(bind=conn) as session:
# 关键：监听事务结束事件，自动重建 savepoint
@listens_for(session.sync_session, "after_transaction_end")
def restart_savepoint(session, transaction):
if transaction.nested and not transaction._parent.nested:
session.sync_session.begin_nested()
# 覆盖 FastAPI 的 get_session 依赖
app.dependency_overrides[get_session] = lambda: session
try:
yield session
finally:
app.dependency_overrides.clear()
await session.rollback()
⚠️ 注意事项：
- 不要使用
BaseHTTPMiddleware操作request.state来传递会话或用户——它绕过 FastAPI 的依赖解析机制，导致类型提示失效、无法被Depends注入、且难以在测试中模拟；request.state仅适合临时存储（如日志上下文），业务关键对象（如db,user）必须通过Depends显式声明；- 测试中务必调用
app.dependency_overrides.clear()清理覆盖，避免污染其他测试用例；- 若使用 PostgreSQL/MySQL，请启用
SAVEPOINT支持，并确保驱动版本兼容（如asyncpg>=0.29）。
✅ 总结
- ✅ 用全局依赖取代中间件：提升类型安全、可维护性与可测试性；
- ✅ 依赖注入即会话生命周期：
get_db()管理会话创建与关闭，get_current_user()复用该会话完成认证； - ✅ 测试 fixture 统一会话：通过
dependency_overrides+ 嵌套事务 + 事件监听，实现“每测试一个事务”的强隔离； - ❌ 避免在中间件中手动管理
request.state.db_session或request.state.user——这是反模式，易引发资源泄漏与并发问题。
这套方案已在多个中大型 FastAPI 项目中验证，兼顾开发体验、运行效率与测试可靠性。
大量免费API接口：立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游，为数据API应用能力赋能！
