本节目标：把「session 开在哪、事务怎么划、连接池配多大、慢查询从哪来」四件事讲成可执行的工程规范，并给出识别 N+1 的实测方法。
适用版本：Python 3.12+（实测 3.14.6）；SQLAlchemy 2.1.4、aiosqlite 0.22.1、greenlet 3.5.6
6.2 事务、连接池与 N+1 治理
6.1 节让模型能建能查，但那些代码都跑在「一次脚本、一个 session」的理想环境里。真实后端要回答的是运行时问题：请求进来时 session 从哪来、什么时候提交、出错怎么回滚、并发上来连接够不够、列表接口为什么悄悄发了 100 条 SQL。这一节逐个拆。
6.2.1 session 生命周期
Session 不是连接，它是工作单元（Unit of Work）：内部维护着身份映射和待提交的变更集，只在需要时才从连接池借一条连接。工程上的铁律是一次请求一个 session，请求结束必关闭。
from sqlalchemy.orm import Session
with Session(engine) as session: # 进入上下文即开始工作单元
book = session.get(Book, 1)
book.price = 39
session.commit() # 提交后连接归还池
# 退出 with：无论是否异常，session 都会 close
Session(engine) 的 with 块只保证 close()，不保证自动提交。忘了 commit() 的后果是改动静默丢失，这是最常见的「改了但没生效」来源。
在 FastAPI 里通常包成依赖，把 session 绑到请求：
from collections.abc import Iterator
from fastapi import Depends
def get_session() -> Iterator[Session]:
with Session(engine) as session:
yield session
@app.get("/books/{book_id}")
def read_book(book_id: int, session: Session = Depends(get_session)):
return session.get(Book, book_id)
yield 之后的清理逻辑在请求结束时执行，正好对应 5.2 节讲的请求上下文。
6.2.2 事务边界
事务边界应该划在业务操作上，而不是每条 SQL 上。需要「要么全成、要么全不成」时用显式事务块：
with Session(engine) as session:
with session.begin(): # 进入即 BEGIN，正常退出 COMMIT，异常 ROLLBACK
src = session.get(Account, 1)
dst = session.get(Account, 2)
src.balance -= 50
dst.balance += 50
实测转账场景（SQLite 内存库）：
in-tx balance: 50
after rollback: 100
after begin(): 42
第一行说明事务内的改动对当前 session 可见；第二行是 session.rollback() 之后数据库值回到 100；第三行是 with session.begin() 正常退出后的持久化结果 42。三种路径——提交、回滚、显式块——覆盖了绝大多数写法。
要点：
- 别在事务里做网络请求或文件 IO，长事务会占着连接、放大锁竞争。
autocommit思维要丢掉：2.x 默认不开自动提交，写操作后必须显式commit()。- 批量插入用
flush()拿主键、commit()收尾，不要在循环里逐条提交。
6.2.3 连接池参数
引擎默认用 QueuePool，参数直接决定并发上限。对一个 SQLite 文件库显式配置并观察状态：
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
engine = create_engine(
"sqlite+pysqlite:///app.db",
poolclass=QueuePool,
pool_size=2, # 常驻连接数
max_overflow=3, # 峰值可临时超出 pool_size 的连接数
pool_pre_ping=True, # 借出前 ping 一次，剔除失效连接
)
实测状态输出（借出 5 条再归还）：
pool class: QueuePool
pool_size: 2 max_overflow: 3
checked out 5: Pool size: 2 Connections in pool: 0 Current Overflow: 3 Current Checked out connections: 5
after 1 close: Pool size: 2 Connections in pool: 1 Current Overflow: 3 Current Checked out connections: 4
after all close: Pool size: 2 Connections in pool: 2 Current Overflow: 0 Current Checked out connections: 0
数字印证了两件事：峰值连接数 = pool_size + max_overflow = 5；归还后常驻连接回到 pool_size = 2，溢出连接被真正关闭而不是囤着。
| 参数 | 含义 | 建议 |
|---|---|---|
pool_size | 常驻连接数 | 按「实例数 × 工作线程/协程」估算 |
max_overflow | 峰值额外连接 | 给突发留余量，别设太大 |
pool_pre_ping | 借出前探活 | 生产务必开，避免用到被服务端断开的连接 |
pool_recycle | 连接最大存活秒数 | 小于数据库的 wait_timeout，防止被静默断开 |
pool_timeout | 等待连接的秒数 | 超时抛 TimeoutError，暴露池不够用 |
一个常被忽略的乘法：总连接数 = 应用实例数 × (pool_size + max_overflow)，必须小于数据库的 max_connections，否则压测时会出现「连不上数据库」。
6.2.4 N+1 的识别
N+1 是指：查主表 1 条 SQL，然后在循环里逐个访问关系字段，又发了 N 条 SQL。识别它最可靠的办法不是读代码，而是数 SQL 条数。用 SQLAlchemy 的事件钩子挂一个计数器：
from sqlalchemy import event
counter = {"n": 0}
@event.listens_for(engine, "before_cursor_execute")
def _count(conn, cursor, statement, params, context, executemany):
counter["n"] += 1
在 10 个作者、每个作者 5 本书（共 50 本）的数据集上跑三种策略：
def run(label, stmt):
with Session(engine) as session:
counter["n"] = 0
authors = session.scalars(stmt).unique().all()
for a in authors:
_ = [b.title for b in a.books] # 触发关系访问
print(f"{label:22} authors={len(authors):2d} "
f"books={sum(len(a.books) for a in authors):3d} SQL={counter['n']}")
实测输出（SQLAlchemy 2.1.4 / SQLite）：
lazy (default) authors=10 books= 50 SQL=11
selectinload authors=10 books= 50 SQL=2
joinedload authors=10 books= 50 SQL=1
惰性加载 11 条（1 条作者 + 10 条每人一次）；selectinload 降到 2 条；joinedload 压到 1 条。数据量翻倍时，惰性那条会线性涨，另两个基本不变——这就是 N+1 的真实代价。生产环境里把这套计数挂到慢请求日志上，比事后翻 slow query 更早发现问题。
6.2.5 selectinload 与 joinedload 的选择
看 selectinload 实际发出的两条 SQL：
SELECT author.id, author.name FROM author
SELECT book.author_id, book.id, book.title FROM book WHERE book.author_id IN (?)
它是「主查询 + 一条 IN 查询」，结果行不会因 JOIN 而膨胀。joinedload 用一条 LEFT OUTER JOIN 搞定，但一对多时会产生笛卡尔积式的重复行，必须 .unique() 去重，而且多层嵌套 JOIN 会让单条 SQL 迅速变复杂。
选择原则：
| 关系 | 推荐 | 原因 |
|---|---|---|
| 一对多 / 多对多（列表） | selectinload | 无行膨胀，SQL 简单 |
| 多对一 / 一对一 | joinedload | 一条 SQL，无膨胀 |
| 深层嵌套 | selectinload 链式 | 每层一条，可控 |
| 明确只取单个对象 | 惰性即可 | 少一次查询 |
也可以把策略写进模型，让默认加载更合理：
class Author(Base):
__tablename__ = "author"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
books: Mapped[list["Book"]] = relationship(
back_populates="author",
lazy="selectin", # 默认就预加载，杜绝遗漏
)
lazy="selectin" 是把「团队约定」编码进模型的常见做法，代价是即使某处用不到关系也会多一条查询，要权衡后使用。
6.2.6 异步 session 与异步 N+1 治理
FastAPI 的 async def 端点里，同步的 Session 会阻塞事件循环，正解是 AsyncSession 配异步驱动 aiosqlite。本机实测环境：SQLAlchemy 2.1.4 + aiosqlite 0.22.1 + greenlet 3.5.6（AsyncSession 依赖 greenlet 提供协程切换）。
from sqlalchemy.ext.asyncio import (
AsyncSession, async_sessionmaker, create_async_engine,
)
async_engine = create_async_engine("sqlite+aiosqlite:///:memory:")
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)
# 建表要经 run_sync 桥接同步的 metadata API
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with AsyncSessionLocal() as session:
session.add_all([User(name="实战"), User(name="入门")])
await session.commit()
users = (await session.scalars(select(User).order_by(User.id))).all()
print("AsyncSession OK ->", [(u.id, u.name) for u in users])
实测输出：
AsyncSession OK -> [(1, '实战'), (2, '入门')]
异步下的 N+1 治理和同步同理，用 selectinload 预加载即可，只是结果要 await：
rows = (await session.scalars(
select(Author).options(selectinload(Author.books))
)).all()
在同一份 10 作者 / 50 本书的数据上，把事件钩子挂到 async_engine.sync_engine 上计数：
async selectinload: authors=10 books=50 SQL=2
与同步版完全一致——预加载把 11 条 SQL 压到 2 条，换成异步不会改变这个结论。
两个异步专属的坑：
- 惰性加载会抛
MissingGreenlet：AsyncSession不能在属性访问时隐式发 IO，关系必须显式selectinload/joinedload，或用await session.refresh(obj, ["books"])补加载。 expire_on_commit=False几乎必开：默认True时，commit()后访问对象属性会触发一次刷新查询，在异步里同样容易踩MissingGreenlet。
驱动层也可以直接使用 aiosqlite 0.22.1（不经过 SQLAlchemy），适合轻量脚本：
import asyncio, aiosqlite
async def main() -> None:
async with aiosqlite.connect(":memory:") as db:
await db.execute("CREATE TABLE book (id INTEGER PRIMARY KEY, title TEXT)")
await db.executemany(
"INSERT INTO book (title) VALUES (?)",
[("Py 实战",), ("Py 入门",)],
)
await db.commit()
async with db.execute("SELECT id, title FROM book ORDER BY id") as cur:
print(await cur.fetchall())
asyncio.run(main())
实测输出：
[(1, 'Py 实战'), (2, 'Py 入门')]
aiosqlite OK
本机无 PostgreSQL 服务端，所有示例均用 SQLite + aiosqlite 实测；PostgreSQL 专属方言（JSONB、SERIAL、ON CONFLICT 等）未实测。
延伸阅读：Python 数据库与 ORM 完全指南 有连接池与隔离级别的更完整对照。
小结
Session是工作单元不是连接，铁律是「一次请求一个 session，结束必关闭」；with Session()只保证close，不保证commit。- 事务边界划在业务操作上，用
with session.begin()显式块；别在事务里做网络或文件 IO。 - 连接池峰值 =
pool_size + max_overflow，生产必开pool_pre_ping；总连接数要小于数据库的max_connections。 - N+1 用事件钩子数 SQL 条数来识别：本机 10 作者 50 本书，惰性 11 条、
selectinload2 条、joinedload1 条。 - 列表类关系优先
selectinload（无行膨胀），多对一用joinedload（单条 SQL）；也可用lazy="selectin"把策略写进模型。 AsyncSession（SQLAlchemy 2.1.4 + aiosqlite 0.22.1 + greenlet 3.5.6）实测可用；异步下关系必须显式预加载，否则报MissingGreenlet，且expire_on_commit=False几乎必开。
模型和运行时都稳了，但表结构还会变——加列、改类型、回填数据。这些变更怎么在不丢数据的前提下逐步应用、随时回滚？下一节的 Alembic 迁移给出答案。
阅读导航：上一节：SQLAlchemy 2.x ORM 与类型化模型 · 下一节：Alembic 迁移与数据演进 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。