FastAPI依赖中不能直接使用asynccontextmanager,必须封装为async def函数并await其调用结果,或改用原生async def依赖配合yield实现生命周期管理。

FastAPI 依赖中直接使用 asynccontextmanager 会报错
FastAPI 的依赖注入系统默认不识别 asynccontextmanager 装饰的函数——它会把返回的 AsyncGenerator 当作普通异步可调用对象,但不会自动执行 __aenter__/__aexit__。你可能会看到类似错误:TypeError: async generator cannot be awaited 或依赖返回值是 而非实际资源。
必须用 Depends 包裹并显式 await 异步生成器
核心解法是:把 asynccontextmanager 函数封装成一个普通 async def 依赖函数,并在其中 await 它的调用结果(即进入上下文)。FastAPI 会正确 await 这个依赖函数,从而触发资源初始化与清理。
-
asynccontextmanager本身不能直接传给Depends(),必须“展开”一层 - 不要写
Depends(my_async_cm),要写Depends(_wrap_my_async_cm) - 包裹函数里必须用
async with或手动await cm.__aenter__(),否则资源不生效
from contextlib import asynccontextmanager from fastapi import Depends, FastAPI@asynccontextmanager async def db_session(): session = await get_db_session() # 假设这是异步创建 session try: yield session finally: await session.close()
✅ 正确:包装为 async def 依赖函数
async def get_db(depends_db_session=Depends(db_session)): async for session in depends_db_session: # 注意:这是唯一能安全遍历的方式 yield session
aexit 会在生成器退出时自动触发
app = FastAPI() @app.get("/items") def read_items(db=Depends(get_db)): return {"status": "ok"}
更推荐:改用原生
async def 依赖 + 手动生命周期管理比起绕一圈用
asynccontextmanager,FastAPI 更自然地支持纯async def依赖函数——你完全可控何时初始化、何时清理,且语义清晰,调试友好。
- 初始化逻辑写在
yield之前,清理逻辑写在try/finally中 - 避免了
asynccontextmanager和Depends叠加导致的嵌套 generator 类型混淆 - 所有中间件、路由、子依赖都能正常 await 它,无兼容性风险
async def get_db(): session = await get_db_session() try: yield session finally: await session.close()@app.get("/items") def read_items(db=Depends(get_db)): # FastAPI 自动识别 async generator 依赖 return {"db_id": id(db)}
注意
yield 后的清理时机和异常传播FastAPI 对
async def依赖中yield的处理是:在请求结束(或依赖链中断)时,恢复协程并执行yield后代码。这意味着:
- 如果
yield后的清理代码抛出异常,它**不会**被上层捕获,可能静默失败或影响后续请求(尤其在连接池场景) - 若想确保清理一定发生,建议在
finally块中做,并对关键操作加日志或重试逻辑 - 不要在
yield后依赖当前 request 状态(如request.state),因为此时 request 已结束
asynccontextmanager 是 Python 标准工具,但 FastAPI 的依赖模型只认两种东西:同步 callable 或 async generator。硬塞第三种类型,就得自己桥接——而桥接点恰恰是最容易漏掉 await 或误判生命周期的地方。










