FastAPI + SQLAlchemy 集成坑：session 生命周期、懒加载与连接池

一句话先说结论：FastAPI + SQLAlchemy 的坑大多出在“session 到底归谁管”上。正确姿势是每个请求一个 session，用 Depends(get_db) 的 yield 依赖配 finally 保证关闭、异常时回滚；同时别在 async def 路由里直接跑同步 Session（会阻塞事件循环），也别在 session 关闭后再碰懒加载属性。

背景

SQLAlchemy 的 Session 是个有状态的会话：它内部有个 identity map（一级缓存），负责攒改动、commit 提交、close 释放连接回连接池。它被设计成“短生命周期”，最理想是一请求一 session，用完即关。一旦这个生命周期和 FastAPI 的请求/依赖生命周期错位，泄漏、脏数据、用已关闭的 session 之类的问题就都冒出来了。

现象

三种典型症状：

重启应用后连接池慢慢用光，或数据库连接数直线上升——session 没关，泄漏了。
某次 commit 之后再去读对象的关联属性，报 DetachedInstanceError 之类——对象被“detach”了，还在读它的懒加载属性。
async def 路由里用同步 SQLAlchemy Session.query()，压测时吞吐上不去、请求排队——数据库这块正在把事件循环拖死。

根因分析

坑 1：session 没走 yield 依赖，生命周期失控

FastAPI 官方推荐的写法是 yield 依赖，把 session 的出生和死亡绑定在请求上：

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()   # 响应发出后执行，保证一定关

yield 之后的代码在响应返回后才跑，finally 保证即使路由抛异常也会 close。而很多项目图省事，在路由函数体里直接 db = SessionLocal()，或者为了 streaming / Celery / 后台任务在 DI 框架外另开 session——这些 session 没人保证关闭，就成了泄漏点。更糟的是“跨生命周期复用”：后台任务拿到的 session 可能已经在别处被关闭，触发 use-after-close。

坑 2：async 路由里跑同步 Session

这其实是第一篇「async def 里别跑同步代码」的数据库特化版：同步 SQLAlchemy 走的是同步驱动，session.query(...).all() 不会让出事件循环，整个 loop 被卡到查询返回。解法见下。

坑 3：懒加载与 detach

commit 默认会把 session 里对象的属性标记为 expired，下次访问会重新去 DB 取——但此时如果 session 已经 close，就没法取了，抛异常（未配置 expire_on_commit=False 时尤其常见）。这也是为什么从 ORM 对象直接序列化返回给接口时，容易在返回阶段炸掉。

解决方案

方案 A：yield 依赖 + 手动 commit/rollback 语义

把“请求级事务”做实，推荐这么写：

from typing import Generator

def get_db() -> Generator:
    db = SessionLocal()
    try:
        yield db
        db.commit()          # 成功才提交
    except Exception:
        db.rollback()        # 异常回滚
        raise
    finally:
        db.close()           # 一定关闭

路由里用 Depends(get_db) 注入，绝不再手动 SessionLocal()。

方案 B：改用异步 SQLAlchemy（真正的解法）

如果接口是 async def 且要跑在高并发下，就用 SQLAlchemy 2.0 的异步支持 + asyncpg：

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine("postgresql+asyncpg://user:pwd@localhost/db", pool_pre_ping=True)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users/{uid}")
async def get_user(uid: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == uid))
    return result.scalar_one_or_none()

expire_on_commit=False 顺手解决了 commit 后属性过期导致的 detach 异常。注意 MySQL 要换 aiomysql/asyncmy 驱动的异步 URL。

方案 C：不想迁移，就把阻塞 ORM 丢线程池

要么把路由改成同步 def（FastAPI 自动丢线程池）：

@app.get("/users/{uid}")
def get_user(uid: int, db: Session = Depends(get_db)):   # 同步 def
    return db.query(User).filter(User.id == uid).first()  # 在线程池跑，不阻塞 loop

要么用 run_in_threadpool 手动包装：

from fastapi.concurrency import run_in_threadpool

@app.get("/users/{uid}")
async def get_user(uid: int):
    user = await run_in_threadpool(query_user, uid)
    return user

连接池别忘配

连接池大小要和 worker 并发匹配，否则高并发下“同步模式开 100 线程要 100 个连接”会把数据库打爆。异步模式单线程用有限连接即可跑满 CPU。pool_pre_ping=True 做心跳保活，避免拿到失效连接。

小结

铁律：一请求一 session，走 yield 依赖 + finally close，别在 DI 外开 session。
async 路由配 async SQLAlchemy（AsyncSession + asyncpg），同步路由或 run_in_threadpool 配同步 Session。
expire_on_commit=False 避开 commit 后的 detach 懒加载坑。
连接池大小、pool_pre_ping 这些生产参数别用默认一走了之。