FastAPI项目SQLAlchemy会话工厂搭建与依赖注入实战指南

1. FastAPI项目为什么需要一套会话工厂

如果你的FastAPI项目只是几个接口、自己随手连个数据库，那确实不需要专门去搭一套会话工厂，直接在每个路由函数里创建连接、用完关闭就行。但只要项目规模一上来，接口多了、并发上来了、团队里好几个人一起维护了，你一定会遇到几个绕不开的问题：连接怎么复用、事务边界在哪、什么时候提交什么时候回滚、每个请求的数据库会话是不是同一个。这些问题如果靠每个人自己约定，几乎百分之百会在某个夜里被线上报错叫醒。

我最早把FastAPI接到生产环境时，就没认真做会话管理，直接在路由里SessionLocal()一把梭。结果接口一压测，报了一堆连接泄漏，数据库连接池被打满，日志里全是QueuePool limit of size。后来老老实实把所有数据库访问收敛到一套会话工厂，再配合FastAPI的依赖注入把会话生命周期管起来，问题才彻底解决。所以今天这篇就从“为什么需要会话工厂”开始，把这套东西一步步搭出来，顺带把我踩过的坑都写清楚。

1.1 会话工厂到底在解决什么问题

先想清楚一件事：SQLAlchemy里的Session不是数据库连接，它是连接之上的一个工作单元。会话负责跟踪你操作过的对象、缓存状态、管理事务，到提交时才真正把改动写进数据库。你如果哪个接口临时create_engine、再顺手建个Session，代码确实能跑，但代价是：

• 每来一个请求就初始化一遍引擎和会话，资源开销极大；
• 事务边界全靠人体自觉，谁忘了commit，改了半天的数据就没了；
• 会话随手丢弃不关闭，连接池很快就会被打满；
• 团队里每个人写一套风格，排查问题成本翻倍。

会话工厂要做的事，就是把这些“每个人都要自己操心”的事情集中收口。工厂负责创建会话，请求进来我们从中拿一个会话使用，请求结束统一关闭；事务的提交与回滚也放到统一的位置去处理。这样业务代码里只需要关心“我怎么查、怎么改”，不需要知道会话怎么来的、怎么没的。

1.2 一个典型的反例

先给一个反面示例，方便对照理解。下面这种写法看起来很直接，但实际上是埋雷。

# 反例：每个请求都临时创建引擎和会话 from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker def get_orders(): engine = create_engine("postgresql://user:pass@localhost/dbname") Session = sessionmaker(bind=engine) session = Session() try: return session.execute("select * from orders").fetchall() finally: session.close()

每调用一次get_orders，就重新走一遍连接创建、引擎初始化、会话建立，完后关掉再走一次销毁流程。本地单线程时问题不明显，一旦并发上来，数据库端就会感受到大量短连接反复建立销毁，应用层也白白耗掉不少时间。

把会话工厂抽出来之后，引擎全局只建一次，连接复用连接池，会话按需创建、用完即关。业务层只看到干净的Session对象，代码整洁性和性能都会好很多。

2. 搭建会话工厂前的准备：选型和目录规划

会话工厂本身不是一个复杂的东西，但它在项目里的位置很重要。它既是基础设施，又是所有数据库操作的入口，所以一开始得把依赖选型和目录结构定好，否则写一半回头再改，牵扯的面会很大。

2.1 依赖选型：同步还是异步

FastAPI天然支持异步，但数据库访问这块要说清楚：SQLAlchemy本身有同步的Session，也有异步的AsyncSession。做选型时主要看你的项目风格和团队熟悉度。

• 如果你在接口里用了def声明同步路由，那用同步Session最简单，几乎不增加额外学习成本。
• 如果你的路由都是async def，并且希望数据库操作也不阻塞事件循环，那就用AsyncSession，配合asyncpg之类异步驱动。
• 如果你的项目还在用旧版SQLAlchemy 1.4，也能跑，但建议直接上SQLAlchemy 2.0，API更统一，异步支持也已经很成熟。

我自己现在的习惯是：新项目直接上SQLAlchemy 2.0 + asyncpg +AsyncSession。但要注意一点，异步会话虽然在IO上不阻塞，但如果你在async def路由里调用了某个同步的数据库工具函数，事件循环一样会被卡住。所以异步不只是换个类名那么简单，整条链路都得是异步的。

如果你刚入门，建议先用同步版本把会话工厂的概念吃透，然后再切换到异步。这篇后面我会把同步和异步的代码都写出来，你按需取舍。

2.2 目录结构建议

会话工厂不必单独拆成一个包，但至少要在项目里有一处集中的位置。一个常见、也足够干净的结构是这样的：

app/ ├─ main.py # FastAPI入口，挂载路由和依赖 ├─ config.py # 数据库配置、环境变量读取 ├─ db/ │ ├─ __init__.py │ ├─ base.py # DeclarativeBase 基类，供模型继承 │ ├─ session.py # 引擎、会话工厂、get_db依赖 │ └─ models.py # 你的 ORM 模型 └─ routers/ └─ order.py # 业务路由

把session.py单独放在db包里，之后所有跟会话有关的东西都从这儿导入，不要在业务代码里自己create_engine。这个约定虽然简单，但长期维护下来价值非常大。

配置项建议放到环境变量里，比如DATABASE_URL，然后交给pydantic-settings或os.getenv读取。不要把连接串硬编码在代码里。连接串里通常包含用户名、密码、地址，硬编码意味着代码仓库里的每个人都有权限看到生产数据库地址，这本身就是个安全隐患。

3. 会话工厂的三个核心环节实现

现在就进入正题：一套可用的会话工厂，至少要包含引擎创建、会话工厂创建、FastAPI依赖注入三个环节。这也是整个方案的骨架。

3.1 第一步：数据库引擎

数据库引擎是SQLAlchemy与数据库打交道的底层入口，负责维护连接池、解析方言、执行SQL。引擎在进程生命周期里应该只创建一次，所以通常放在模块级别，或者在FastAPI的lifespan里初始化。

# app/db/session.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase DATABASE_URL = "postgresql://user:password@localhost:5432/appdb" engine = create_engine( DATABASE_URL, pool_size=10, # 连接池保持的连接数 max_overflow=20, # 超过 pool_size 后最多还能新建多少个连接 pool_pre_ping=True, # 取出连接前先探活，避免拿到的连接已断开 echo=False, # 生产环境别开，调试SQL才打开 )

这里三个参数影响很大。

pool_pre_ping=True必须开。数据库偶尔会主动断掉空闲连接，如果不开探活，应用可能拿到一个已经失效的连接，然后报出各种莫名其妙的错误。开了之后，SQLAlchemy取出连接时会先执行一次轻量查询确认连接可用，有问题就自动换一条，对业务完全透明。

pool_size和max_overflow要根据并发量估算。假设你的服务单实例最大并发请求是100，每个请求一个会话，同一时刻最多需要100个连接。那pool_size可以设成20，max_overflow设成80，保证平时频繁使用的连接被缓存，突发流量时也能临时扩容。设置得太大反而没有意义，因为数据库服务端也有自己的连接数上限。

3.2 第二步：会话工厂

会话工厂由sessionmaker创建，它的作用就是按照预设参数批量生成Session实例。

SessionLocal = sessionmaker( bind=engine, autoflush=False, expire_on_commit=False, )

两个参数很多人不注意，但直接影响行为：

autoflush=False建议设置。autoflush默认是True，意思是查询前SQLAlchemy会先把尚未提交的改动flush到数据库。看起来方便，但容易在事务中间产生意外的数据库写操作，造成行锁和性能问题。设成False之后，什么时候写数据库由你自己控制，行为更可预测。

expire_on_commit=False也要设。默认情况下，一旦commit，会话里所有对象的属性会过期，之后你再访问order.status，SQLAlchemy会再发一条SQL去数据库查一遍。设成False，提交后对象还能保留之前的属性值，可以在事务结束后继续安全读取。

如果用的是异步模式，代码几乎一样，只是换几个类名：

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession engine = create_async_engine( "postgresql+asyncpg://user:password@localhost:5432/appdb", pool_size=10, max_overflow=20, pool_pre_ping=True, ) AsyncSessionLocal = async_sessionmaker( bind=engine, autoflush=False, expire_on_commit=False, )

3.3 第三步：FastAPI依赖注入管生命周期

会话工厂建好之后，最难的一步是怎么保证每个请求都用独立会话，并且请求结束一定关闭。直接在每个路由里手动SessionLocal()有点繁琐，还容易漏关闭。FastAPI的依赖注入在这里就非常顺手了。

from fastapi import Depends from sqlalchemy.orm import Session def get_db(): db = SessionLocal() try: yield db finally: db.close()

然后在路由里这样用：

from fastapi import APIRouter, Depends from sqlalchemy.orm import Session router = APIRouter() @router.get("/orders/{order_id}") def get_order(order_id: int, db: Session = Depends(get_db)): order = db.get(Order, order_id) return {"order": order}

关键在于yield和finally。FastAPI会在请求进来时调用get_db，拿到db传给接口，整个请求期间路由函数复用这一个会话；请求结束时，FastAPI会回到生成器继续执行，走入finally把会话关闭。就算接口里抛了异常，finally也会执行，会话不会泄漏。

异步版依赖写法换成AsyncSession和async def即可：

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

不过这里有个细节，如果依赖是async def，那它的yield部分会在FastAPI的异步上下文中执行，路由最好也保持async def。我之前经历过同步路由注入异步会话，代码能跑，但总感觉别扭，后来统一成异步才理顺。

3.4 为什么用依赖注入而不是用全局变量

有人可能问，直接把Session做成全局单例不就行了？确实有很多项目这么干，但这是另一种坑。

Session不是线程安全的，它在内部持有事务状态。假设全局只有一个Session，两个请求同时往里面写数据，事务状态会互相污染，轻则数据错乱，重则直接报错。FastAPI是并发框架，同一时刻进程内可能在处理大量请求，全局单例Session就是定时炸弹。

依赖注入的好处是：每个请求拿到独占的会话实例，互不干扰，生命周期又跟着请求自动走。这才是“会话工厂”这个方案真正值钱的地方——工厂提供统一标准，DI框架负责按需分发和回收。

4. 业务代码里怎么正确使用会话工厂

会话工厂和依赖注入搭好之后，接下来的重点就是业务代码的写法。这里有几个常见的决策场景，如果处理不好，就算工厂搭得再漂亮，也会在日常使用中翻车。

4.1 查询、写入、提交与回滚

最常见的使用模式是这样的：

@router.post("/orders") def create_order(payload: OrderCreate, db: Session = Depends(get_db)): order = Order(**payload.dict()) db.add(order) db.commit() db.refresh(order) return {"id": order.id}

这里commit之后调用refresh，目的是把数据库生成的默认值、自增ID等回填到内存对象里。因为你设置了expire_on_commit=False，如果不refresh，order.id可能还是空值，直接返回就有问题。

写入操作要注意异常处理。如果commit失败，不主动回滚，当前会话可能处于“部分写入但事务已损坏”的状态，继续复用这个会话很容易出问题。稳妥的写法是包一层try/except：

from fastapi import HTTPException @router.post("/orders") def create_order(payload: OrderCreate, db: Session = Depends(get_db)): order = Order(**payload.dict()) db.add(order) try: db.commit() db.refresh(order) except Exception: db.rollback() raise HTTPException(status_code=500, detail="写入失败") return {"id": order.id}

注意，rollback之后当前事务就作废了，但会话本身还能继续用。如果你希望这个请求无论如何都只能有一个干净事务，那可以在异常时让整个依赖链抛出错误，FastAPI会自动处理。

4.2 子依赖与事务边界

有时候业务会拆分多个依赖函数，每个函数都声明db: Session = Depends(get_db)。要理解一点：只要用的是同一个get_db依赖，FastAPI默认会缓存结果，所以同一个请求里所有依赖拿到的都是同一个Session实例。这一点特别重要，不然你可能以为两个函数用的是不同会话，实际却是同一个。

举个例子：

def validate_order(payload: OrderCreate, db: Session = Depends(get_db)): # 做一些校验，读数据库 return True @router.post("/orders") def create_order(payload: OrderCreate, db: Session = Depends(get_db)): validate_order(payload, db) # 这里继续写库

即使不显式从路由传入db，validate_order里的Depends(get_db)拿到的还是同一个会话。这种机制可以确保一个请求内部的数据库操作在同一个事务上下文里，避免跨会话的隐式分布式事务。

但也正因为如此，没搞清楚依赖缓存机制的人会困惑：为什么子依赖里改了查不到，或者为什么两个接口之间会互相影响？其实不是会话串了，而是“同一个请求”内共享了会话，不同请求之间会话才是隔离的。

4.3 什么时候应该使用with上下文管理器

有时候你并不在FastAPI路由里操作数据库，而是在一个独立的后台任务、脚本或者Cron任务里。比如Celery任务需要更新一批数据。这种情况下同样可以用会话工厂，但不用依赖注入，而是用上下文管理器手动管理。

def update_inventory(order_id: int): with SessionLocal() as db: order = db.get(Order, order_id) order.status = "paid" db.commit()

with SessionLocal() as db会在代码块结束时自动关闭会话，比手动close更省心。注意这里不会自动提交，需要自己显式调用commit()。很多初学者以为出了with块数据就保存了，不是的，with只负责关闭连接，是否提交是你自己的事。

5. FastAPI中会话工厂的常见问题与排查

写了几百个接口之后，我总结出几个和会话工厂强相关的典型问题。这些问题不经历一场生产事故，很难真正记住。我把排查思路整理成了一张对照表，按这套逻辑走，定位问题一般不会跑偏。

5.1 连接池耗尽

现象：日志里频繁出现QueuePool limit of size ... overflow ... reached，接口响应变慢，数据库CPU却不高。

原因：会话泄漏。大概率是某个分支没有关闭会话，比如在依赖里写了yield db，但绕过了finally，或者在业务代码里手动SessionLocal()却忘了close。连接一直占着不还，连接池迟早被打满。

排查步骤：

1. 检查所有SessionLocal()是否都包裹在with块或者try/finally中。
2. 检查依赖生成器是否都用了yield配合finally。
3. 用工具查看数据库端活跃连接数，和应用端连接池使用情况对比。
4. 临时把pool_size调大只是缓解症状，根治还是要找到哪里泄漏。

5.2DetachedInstanceError异常

现象：接口里对一个从数据库查出来的对象做懒加载，比如访问order.items、order.customer.name，突然报DetachedInstanceError。

原因：对象从会话里“脱离”了。一般是两种情况：一是会话已经关闭，你拿到对象之后又在新会话里尝试访问未加载的属性；二是你手动expunge了对象。

排查思路：如果你需要在会话关闭后继续使用对象，需要在会话内先把需要关联的属性加载完整，比如用selectinload一次性把关联数据查出来。懒加载只适合在会话打开期间使用，不要指望它跨会话生效。

5.3 异步与同步混用

现象：接口用async def，数据库却是同步的，压测时发现事件循环卡顿。

原因：同步数据库操作会阻塞线程。FastAPI的async def路由运行在事件循环里，你在中间调用了同步的db.execute(...)，整个事件循环都会卡住，其他并发请求全部排队。

排查思路：检查你用的SQLAlchemy是同步还是异步。如果你在main.py里用了AsyncSessionLocal，但业务代码里写的是from sqlalchemy.orm import Session，那大概率是同步会话。要统一用AsyncSession，并且数据库驱动也要配成异步的，比如asyncpg。

5.4 提交了但数据没变

现象：db.commit()执行成功，没有任何报错，但数据库里数据没变化。

原因：autoflush=False时，如果提交前没有把改动主动flush，某些情况下SQLAlchemy认为没有待提交的改动。但实际上更常见的原因是你忘记db.add()了。比如你修改了一个从数据库查出来的对象属性，然后直接commit，这个动作本身是有效的，系统会跟踪已加载对象的变化。但如果你新建了一个对象却忘了db.add()，ORM根本不知道它的存在，commit自然什么都不做。

这个坑最容易出现在“从接口参数直接构造对象”的代码里。养成习惯：新增对象后先看一眼是否调用了db.add()，再想commit的事。

这些排查经验如果能提前了解，实际遇到问题时至少不会慌。会话工厂本身不复杂，真正复杂的是用好它之后，你还要对它的行为模型有清晰认知。

6. 一个可以直接抄走的完整模板

如果你不想纠结太多细节，下面这套模板可以直接复制到项目里跑通，然后再按需调整。

6.1 最小可运行代码

# app/db/session.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase DATABASE_URL = "postgresql://user:password@localhost:5432/appdb" engine = create_engine( DATABASE_URL, pool_size=10, max_overflow=20, pool_pre_ping=True, echo=False, ) SessionLocal = sessionmaker( bind=engine, autoflush=False, expire_on_commit=False, ) class Base(DeclarativeBase): pass

# app/db/models.py from sqlalchemy import String, Integer from sqlalchemy.orm import Mapped, mapped_column from .session import Base class Order(Base): __tablename__ = "orders" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) status: Mapped[str] = mapped_column(String(32), default="pending")

# app/db/deps.py from .session import SessionLocal def get_db(): db = SessionLocal() try: yield db finally: db.close()

# app/main.py from fastapi import FastAPI, Depends from sqlalchemy.orm import Session from .db.deps import get_db from .db.models import Order app = FastAPI() @app.get("/orders/{order_id}") def get_order(order_id: int, db: Session = Depends(get_db)): return db.get(Order, order_id)

如果你项目用的是异步模式，直接把sessionmaker换成async_sessionmaker，def get_db换成async def get_db，路由也需要改成async def。数据库驱动记得用asyncpg。

6.2 使用这套模板的约定

这个模板能跑，但能跑不等于万事大吉。想用得长久，还需要团队里立下几个约定：

• 所有数据库会话必须通过get_db依赖获取，禁止在路由里直接SessionLocal()。
• 所有写入操作必须显式提交，禁止依赖自动提交（SQLAlchemy默认就没有自动提交，但有人会误以为with块会提交）。
• 不要在会话关闭后访问懒加载属性，需要提前用selectinload、joinedload加载。
• 不要在异步路由里执行同步数据库查询，要保持异步链路统一。

这些约定不是教条，而是我吃过亏之后总结出的底线。踩过连接池告警、被DetachedInstanceError咬过几次、体验过异步阻塞后，你就会理解每一条背后都有一次真实的线上经历。

7. 进阶扩展：会话工厂还能往哪个方向演化

如果你的项目已经稳定运行，基本会话工厂用顺手了，还可以往三个方向继续演进。

第一个方向是加超时控制。FastAPI依赖本身可以挂在timeout参数上，让整个请求最多执行多久。数据库层也可以给会话绑定超时，比如connect_args={"connect_timeout": 5}，避免数据库连不上时应用长时间傻等。

第二个方向是分库分表或读写分离。单会话工厂绑定的只是一个数据源，如果你的项目开始有读写分离需求，可以把会话工厂做成可配置的，根据上下文选择走主库还是从库。实现上通常要拆成两个引擎、两个会话工厂，然后按业务路由指定使用哪个。

第三个方向是接入OpenTelemetry这类可观测体系。在SessionLocal外面套一层自定义包装，每次会话创建、提交、回滚、关闭都记录耗时和状态。这样后续排查接口慢的时候，可以直接看到是哪一步数据库操作拖慢了整个请求。

这些扩展都是后话，前提始终是先把基础的会话生命周期管理好。工厂建了不乱用、用了不忘关、关了不再摸，这套基本功扎实了，什么花活都能接得住。

最后分享一个我个人长期在用的习惯：每次写完新接口，都会在本地开一条压测命令，模拟二三十个并发请求连续跑几分钟，然后盯着数据库端连接数曲线。如果连接数一路涨上去不回落，基本可以断定会话有泄漏。这个检查动作成本很低，但能帮你提前发现很多生产环境下才会暴雷的问题。