应注入会话工厂或上下文管理器而非session实例,因直接返回session()会导致多请求共享、状态污染、事务混乱及连接泄漏;须用yield配合try/finally或async with确保fastapi控制生命周期并自动清理资源。

直接用 Depends 注入数据库会话本身是错的——你真正该注入的是“会话工厂”或“上下文管理器”,而不是一个已创建的 Session 实例。否则每次请求都会复用同一个会话,导致状态污染、事务混乱甚至连接泄漏。
为什么不能在依赖函数里直接 return Session()
常见错误写法:
def get_db():
return SessionLocal() # ❌ 错!返回实例,不是工厂
问题很实在:
- 这个
Session实例被多个请求共享,commit()/rollback()互相干扰 - 没有自动关闭机制,
db.close()容易漏写,连接池慢慢耗尽 - 异常路径下资源无法释放,日志里频繁出现
ConnectionResetError或TimeoutError
正确做法:用 yield + async with(同步/异步都要覆盖)
核心原则:让 FastAPI 控制生命周期,而不是你手动 new/close。
同步场景(SQLAlchemy ORM):
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
异步场景(SQLAlchemy 2.0+ asyncpg):
async def get_db():
async with async_session_maker() as session:
yield session
关键点:
-
yield前是“准备”,yield后是“清理”,FastAPI 保证无论路由是否抛异常都会执行清理段 - 同步用
try/finally,异步必须用async with,不能混用 - 不要在依赖函数里做
session.commit()—— 事务边界应在路由或 service 层显式控制
避免嵌套依赖中意外复用同一会话
当多个依赖都调用 get_db(),比如:
def get_user(db: Session = Depends(get_db)):
return db.query(User).first()
<p>def get_order(db: Session = Depends(get_db)):
return db.query(Order).first()</p><p>@app.get("/summary")
def get_summary(user=Depends(get_user), order=Depends(get_order)):
...
</p>
这会导致同一个请求内 get_db() 被调用两次,生成两个独立会话 —— 违反事务一致性。解决方法:
- 把
db提到最外层路由参数,内部函数通过参数传入,不各自Depends - 或者启用 FastAPI 的依赖缓存:在
Depends(get_db, use_cache=True)(默认已开启,但仅限同一请求内) - 更稳妥的是改用类依赖,重载
__call__并加@lru_cache控制单例行为(需配合 request context)
测试时如何替换数据库依赖
单元测试不能连真实 DB,得注入 mock 会话:
# 测试 client app.dependency_overrides[get_db] = lambda: TestingSessionLocal()
注意两点:
- 覆盖必须在
TestClient初始化之后、请求发出之前设置 - 测试结束后记得清除:
app.dependency_overrides.clear(),否则影响后续测试 - 如果依赖用了
yield,mock 函数也得用yield,不然 FastAPI 会卡住等待finally执行
最常被忽略的一点:依赖函数里的 yield 不是语法糖,它是 FastAPI 触发资源清理的唯一钩子。少写一个 finally 或漏掉 await session.close(),线上就可能跑几天后突然报 Too many connections。别信“应该没问题”,检查每个 Depends 的退出路径。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!











