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()，壓測時吞吐上不去、請求排隊——數據庫這塊正在把事件循環拖死。

第 2 種會甩出運行時例外。報錯原文：

sqlalchemy.orm.exc.DetachedInstanceError: Parent instance <User at 0x7f9a1c3e4a30> is not bound to a Session; lazy load operation of attribute 'orders' cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)

完整堆疊：

Traceback (most recent call last):
  File "/app/api/users.py", line 48, in get_users
    return {"name": user.name, "orders": [o.id for o in user.orders]}
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/attributes.py", line 481, in __get__
    return self.impl.get(state, dict_)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/attributes.py", line 941, in get
    value = self._fire_loader_callables(state, key, passive)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/loading.py", line 1603, in load_scalar_attributes
    raise orm_exc.DetachedInstanceError(
sqlalchemy.orm.exc.DetachedInstanceError: Parent instance <User at 0x7f9a1c3e4a30> is not bound to a Session; lazy load operation of attribute 'orders' cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)

同一個根源還有另一個變體 InvalidRequestError，報錯原文：

sqlalchemy.exc.InvalidRequestError: Instance <User at 0x7f9a1c3e4a30> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)

根因分析

坑 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 這些生產參數別用預設一走了之。