SQLAlchemy 异步引擎与会话
FastAPI 是异步框架，数据库这一层自然也得「异步到底」。本章搞定两件事：
FastAPI 是异步框架，数据库这一层自然也得「异步到底」。本章搞定两件事：
- 引擎（engine）——数据库连接的调度中心，怎么建、怎么配；
- 会话（session）——ORM 操作的工作台，怎么开、怎么提交、怎么还。
主线按 PostgreSQL + asyncpg 讲（生产环境的主流选择），配套 demo 用 SQLite + aiosqlite——零安装零配置，clone 下来就能跑。两者对 SQLAlchemy 来说只差一行连接串，其余代码一个字不用改。
本章配套 demo： -
demo/demo_engine.py：异步引擎 + text() 裸 SQL，建表、插入、查询 -demo/demo_session.py：会话工厂 + 三种提交方式对比示例业务统一用「墨集 inkhub 内容平台」：
channel（栏目）和article（文章）两张表。
章节分工提醒：select()/where()/update()/delete() 这些 SQL 表达式（Core 风格）是第 06 章的主场，本章查询一律用最简形式；ORM 模型怎么声明（DeclarativeBase、Mapped）是第 05 章的内容，本章直接借用一个极简模型。
1. 安装：一个方括号引发的血案
作用：装上 SQLAlchemy 的异步能力 + 一个异步数据库驱动。
pip install "sqlalchemy[asyncio]" asyncpg
# 本章 demo 用 SQLite 则装：
pip install "sqlalchemy[asyncio]" aiosqlite
三样东西各司其职：
| 包 | 角色 | 说明 |
|---|---|---|
sqlalchemy |
ORM 框架本体 | 引擎、会话、模型、表达式全在这 |
[asyncio] |
附加依赖（extras） | 方括号语法，会额外装上 greenlet——异步引擎底层靠它做协程切换，不装必报错 |
asyncpg / aiosqlite |
数据库驱动 | 光有框架没有驱动，等于有翻译没有快递员 |
常见坑
pip install sqlalchemy asyncpg少了方括号 → 运行时抛MissingGreenlet或InvalidRequestError: The asyncio extension requires greenlet。记住了：异步用法，方括号不能省。psycopg2是同步驱动，配create_async_engine用不了。异步引擎只认异步驱动：PostgreSQL 用asyncpg，SQLite 用aiosqlite，MySQL 用asyncmy。
2. create_async_engine：先把调度中心建起来
作用：创建异步引擎。引擎是进程级单例，掌管连接池、方言翻译、SQL 日志——所有数据库操作的起点。
签名（挑常用的记）：
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine(
url, # 连接串：数据库必填
echo=False, # 是否打印发出的 SQL（调试神器）
echo_pool=False, # 是否打印连接池的借还日志
pool_size=5, # 池内常驻连接数（PG 等服务型数据库用）
max_overflow=10, # 池满后允许临时加开的连接数
pool_pre_ping=False, # 借出前先发 ping 探活
pool_timeout=30, # 借不到连接时最多等多久（秒）
)
连接串格式——数据库类型+驱动://用户:密码@主机:端口/库名：
# PostgreSQL + asyncpg（本章主线）
DATABASE_URL = "postgresql+asyncpg://inkhub:secret@localhost:5432/inkhub"
# SQLite + aiosqlite（demo 用，三根斜线接相对/绝对路径）
DATABASE_URL = "sqlite+aiosqlite:///./inkhub.db"
# 切换数据库 = 改这一行，业务代码零改动
最小示例——跑通「数据库心跳」：
import asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine(DATABASE_URL, echo=True)
async def main() -> None:
async with engine.connect() as conn: # 借一条连接
result = await conn.execute(text("SELECT 1"))
print(result.scalar()) # 1，活着
await engine.dispose() # 进程收尾：放光所有连接
asyncio.run(main())
常见坑
- 每次请求
create_async_engine一个新引擎——连接池就废了，每个池都是一次性的，连接风暴直接把数据库打挂。正确姿势：模块级建一个，全进程共用。 echo=True只该出现在调试时。开着它，每条 SQL 连参数一起打进日志，线上既吵又可能泄敏感数据。- 连接串里密码有特殊字符（
@、:、/）要 URL 编码，否则解析必乱。可用sqlalchemy.engine.URL.create()逐字段安全拼装。 - SQLite 是嵌入式文件库，没有「池」的概念，
pool_size/max_overflow传了也不起作用；这两位是给 PostgreSQL 这类服务型数据库准备的。
3. 引擎、连接池、驱动：一个柜台三家店
一张图讲清三者关系（记这张图，本章就懂一半了）：
你的异步代码（await session.execute(...)）
│
▼
┌──────────────────────────────┐
│ 引擎 engine │ 柜台：接单、排队、翻译方言，
│ create_async_engine() │ 把 Python 调用变成 SQL
└──────────────┬───────────────┘
│ 借出 / 归还
▼
┌──────────────────────────────┐
│ 连接池 pool │ 充电宝柜子：连接用完不扔、
│ pool_size / max_overflow │ 放回柜子，下一位直接复用
└──────────────┬───────────────┘
│ 持有真正的连接
▼
┌──────────────────────────────┐
│ 驱动 driver │ 线：asyncpg 只会插 PostgreSQL，
│ asyncpg / aiosqlite / ... │ aiosqlite 只会插 SQLite，各认各的口
└──────────────┬───────────────┘
▼
┌────────────┐
│ 数据库服务器 │
└────────────┘
用共享充电宝再念一遍：
- 连接池是柜子：借出（
connect()）、归还（退出async with），避免每次都「造一根新线」（TCP 握手 + 认证，很贵）； - 驱动是那根线：它只负责和特定数据库通信，不懂 SQL 怎么拼、事务怎么管——那是框架的事；
- 引擎是柜子的控制系统：决定借谁、还谁、池子多大、断线的充电宝要不要扔。
常见坑
- 连接池不是越大越好。
pool_size + max_overflow的总连接数要对着数据库的max_connections规划，池子开太大只会让数据库排队排到超时。 - 数据库重启后，池里的旧连接全成了「僵尸」。
pool_pre_ping=True让每次借出前先 ping 一下，僵尸自动换新——代价是每次多一个轻量往返，生产环境建议开。
4. text()：裸 SQL 的正确姿势
作用：把一个 SQL 字符串包装成 SQLAlchemy 认识的语句对象。引擎和连接不直接吃字符串，喂之前必须 text() 一下。
签名：text(sql: str) -> TextClause，配合 :参数名 占位符 + 字典传参。
最小示例：
from sqlalchemy import text
async with engine.connect() as conn:
# 1) 无参数：直接包一层
result = await conn.execute(text("SELECT COUNT(*) FROM article"))
print(result.scalar()) # 单值：scalar()
# 2) 命名参数：占位符 + 字典，框架自动做参数化绑定
result = await conn.execute(
text("SELECT * FROM article WHERE channel_id = :cid"),
{"cid": 1},
)
for row in result.fetchall(): # 多行：fetchall()
print(row.id, row.title) # 按列名取值
# 3) 批量插入：传字典列表，一条语句插多行（executemany）
await conn.execute(
text("INSERT INTO channel (name) VALUES (:name)"),
[{"name": "前沿速递"}, {"name": "深度长文"}],
)
常见坑（头号大坑：SQL 注入）
# ❌ 千万别这样：用户输入直接进 SQL
keyword = "' OR 1=1 --"
await conn.execute(text(f"SELECT * FROM article WHERE title LIKE '%{keyword}%'"))
# 拼出来的 SQL 恒真，全表数据裸奔；换成 DELETE 就是删库
# ✅ 永远用占位符
await conn.execute(
text("SELECT * FROM article WHERE title LIKE :kw"),
{"kw": f"%{keyword}%"}, # 整段被当成纯文本，注入无效
)
原理一句话：参数化绑定让数据库先解析 SQL 骨架、再单独收参数值，值永远不会被当成指令执行。另外两个小坑：text() 忘包（conn.execute("SELECT 1") 直接抛 ArgumentError）；PG 的 % 模糊匹配在 text 里不用像 psycopg2 那样写成 %%，占位符统一 :name 风格。
裸 SQL 灵活但要手拼字符串、无类型检查、换数据库方言可能不兼容——所以第 06 章会请出 SQL 表达式。但在复杂聚合、特调优化的场景，裸 SQL 依然是最后一道底牌。
5. connect() vs begin()：借连接的两种方式
从引擎手里拿连接，两个入口：
| 入口 | 事务行为 | 适用场景 |
|---|---|---|
async with engine.connect() as conn |
借一条连接，不自动提交；要提交得手动 await conn.commit() |
只读查询；想精细控制事务边界 |
async with engine.begin() as conn |
进入即开事务，正常退出自动 commit，异常自动 rollback | 写操作（INSERT/UPDATE/DELETE）的默认选择 |
# 查询：connect 就够
async with engine.connect() as conn:
result = await conn.execute(text("SELECT * FROM channel"))
# 写入：begin 省心，块里挂了自动回滚，整段同生共死
async with engine.begin() as conn:
await conn.execute(
text("INSERT INTO article (title, channel_id) VALUES (:t, :c)"),
{"t": "异步编程入门：事件循环是怎么回事", "c": 1},
)
await conn.execute(
text("UPDATE channel SET name = :n WHERE id = :id"),
{"n": "前沿速递·修订", "id": 1},
)
# 走到这，两条改动一起提交；中间任何一步抛异常，两条一起回滚
常见坑：connect() 里执行了 INSERT 却忘了 commit()，退出上下文只还连接不提交，改动全部蒸发——而且不报错，纯静默丢失，排查起来最气人。
6. async_sessionmaker：会话工厂
引擎和 text() 解决了「能和数据库说话」，但真正干 ORM 活的是会话（session）。会话不能凭空 new，得从工厂里造——这就是 async_sessionmaker。
作用：把「引擎 + 一堆默认配置」打包成一个会话工厂，之后随取随用，配置不散落各处。
签名：
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
session_factory = async_sessionmaker(
engine, # 绑定的引擎
class_=AsyncSession, # 明确用异步会话类（写上，不靠默认）
expire_on_commit=False, # 提交后对象属性是否过期（见第 10 节，FastAPI 场景常关）
autoflush=True, # 查询前自动把挂起的改动 flush 掉
)
最小示例：
# 工厂是进程级单例，建一次用到进程结束
session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
# 用的时候像调用函数一样「造」一个会话
async with session_factory() as session:
...
常见坑
- 旧代码里的
sessionmaker是同步版；异步世界认准async_sessionmaker（从sqlalchemy.ext.asyncio导入）。 - 会话 ≠ 连接。会话是「工作台」，用到数据库时才从引擎的池里临时借一条连接，用完就还。所以「多开会话」比「多开连接」便宜得多。
- 一个会话同一时刻只服务一个协程，别多个请求共享同一个会话（FastAPI 里靠依赖注入保证一人一个，见第 11 节）。
7. 会话上的增删查改
先借用一个极简模型（怎么建的第 05 章细讲，本章只要知道「Channel/Article 是两张表对应的两个 Python 类」即可）：
from sqlalchemy import ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
pass
class Channel(Base):
__tablename__ = "channel"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
class Article(Base):
__tablename__ = "article"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
channel_id: Mapped[int] = mapped_column(ForeignKey("channel.id"))
增：add()
async with session_factory.begin() as session:
article = Article(title="拆解一个电商系统的订单链路", channel_id=2)
session.add(article) # 只是挂到工作台，还没发 SQL
# flush 之后主键自动回填：article.id 有值了
add() 像「把商品放进购物车」——只登记，不结账。SQL 要等 flush/commit（第 9 节）才真正发出。
查：execute() + scalars()
from sqlalchemy import select
async with session_factory() as session:
result = await session.execute(select(Article)) # select 的完整玩法见 06 章
articles = result.scalars().all() # scalars()：剥掉 Row 壳，直接拿模型对象
for a in articles:
print(a.id, a.title)
一句话记住：result 装的是一行行的 Row（像小元组），scalars() 把每行的第一列抽出来——ORM 查询里第一列就是模型对象，所以 scalars().all() 到手的就是 list[Article]。
查一个：scalar_one() / scalar_one_or_none()
article = (
await session.execute(select(Article).where(Article.id == 1))
).scalar_one() # 恰好一行 → Article 对象
| 方法 | 0 行 | 1 行 | 多行 |
|---|---|---|---|
scalar_one() |
抛 NoResultFound |
返回对象 | 抛 MultipleResultsFound |
scalar_one_or_none() |
返回 None |
返回对象 | 抛 MultipleResultsFound |
按主键查、业务上「必须存在」的场景用 scalar_one()，让缺失直接炸出来而不是悄悄传 None 往下走。
改：查出来，直接改属性
async with session_factory.begin() as session:
article = (await session.execute(select(Article).where(Article.id == 1))).scalar_one()
article.title = "异步编程入门：从事件循环到 uvloop" # 改对象 = 购物车里换货
# 提交时 session 自动对比出差异，生成精准的 UPDATE
ORM 的爽点：没有 update() 这个调用。会话记着你查过的对象，改属性即改数据，提交时自动算 UPDATE。
删：delete()
async with session_factory.begin() as session:
article = (await session.execute(select(Article).where(Article.id == 1))).scalar_one()
await session.delete(article) # 注意 await，异步方法
# 提交时自动生成 DELETE
常见坑：session.add() 和 session.delete() 长得像，但前者是同步方法、后者在异步会话里必须 await——漏了 await 删除根本没执行，还不一定报错。
8. 三种提交方式：从手动挡到自动挡
写操作最终都要 commit() 才落库。三种写法，从「全手动」到「全自动」：
方式一：手动挡——commit / rollback / close 全自己管
session = session_factory()
try:
session.add(Article(title="...", channel_id=1))
await session.commit() # 提交
except Exception:
await session.rollback() # 出错回滚
raise
finally:
await session.close() # 无论如何都要关
能看懂就行，日常不建议写——仪式感太重，漏掉任何一个环节都是坑（忘 rollback 占着连接、忘 close 连接泄漏）。
方式二：上下文管关闭——事务仍手动
async with session_factory() as session: # 退出时自动 close
session.add(Article(title="...", channel_id=1))
await session.commit() # 提交还得自己来
async with 只包了「关」，没包「提交」。适合一个会话里干好几摊事、每摊单独提交的精细控制场景。
方式三：自动挡——begin() 一起管
async with session_factory.begin() as session:
session.add(Article(title="...", channel_id=1))
# 正常走到头 → 自动 commit；中途抛异常 → 自动 rollback + close
一行上下文，提交/回滚/关闭全包。日常业务代码的默认选择。
| 方式 | 关闭 | 提交 | 回滚 | 一句话 |
|---|---|---|---|---|
| 手动 | 手动 | 手动 | 手动 | 老式写法，懂原理用 |
factory() |
自动 | 手动 | 手动 | 多事务分段提交 |
factory.begin() |
自动 | 自动 | 自动 | 无脑首选 |
常见坑：方式二里忘了 commit()——退出上下文干干净净、不报任何错，数据没了。这是新手丢数据第一名的原因。
9. flush vs commit：先发 SQL ≠ 先落库
这俩最容易混。用「发文章」打比方：
- flush = 点了「保存草稿」：SQL 真的发到数据库了（INSERT/UPDATE 都执行了），但事务没提交——只有你自己（当前事务）看得见，别人看不见，随时可以反悔；
- commit = 点了「发布」：先自动 flush 所有挂起的改动，再提交事务——从此所有人可见，覆水难收。
async with session_factory.begin() as session:
article = Article(title="本周框架动态", channel_id=1)
session.add(article)
await session.flush() # INSERT 已发出，article.id 已回填
print(article.id) # ✅ 主键能用了（数据库生成的）
# 此刻换一个会话去查这条记录 → 查不到（事务未提交）
# 退出 begin 块，自动 commit，别的会话才看得见
什么时候需要手动 flush？想在提交前拿到数据库生成的主键（比如插入文章后立刻要用 article.id 去建关联记录）。其余时候交给 commit 里的自动 flush 就行。
另外记住 autoflush=True（默认）：每次 session.execute() 查询前，会话会先把挂起的改动 flush 掉——所以你在同一个会话里 add 完马上查，能查到自己的改动，这不是玄学，是 autoflush 在干活。
10. 字段过期与 refresh：commit 之后属性还能读吗
async_sessionmaker 里那个 expire_on_commit，专管这件事。
默认 expire_on_commit=True：提交即「贴封条」。commit 之后，会话把所有关联对象的字段标记为过期——你再一读属性，它就发一条 SELECT 去数据库重新对账。相当于结账后超市把小票上每件商品都盖了「待复核」章，摸一下就得回货架核价。
async with session_factory.begin() as session:
article = Article(title="字段过期演示", channel_id=1)
session.add(article)
await session.flush()
# 会话已关闭，对象脱离会话
print(article.title) # 💥 DetachedInstanceError：想过期属性，但没会话帮你查了
为什么 FastAPI 里几乎都关掉：Web 请求的典型节奏是「commit → 把对象塞进 Pydantic 响应模型 → 返回」，这中间读属性时事务早关了，开着过期只会炸。所以社区惯例：
async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
关掉之后提交完属性照常可读（读的是提交前的内存快照）。
那数据库万一变了呢？用 refresh()——在会话还活着的时候，主动把数据库的最新值同步回对象：
async with session_factory.begin() as session:
article = (await session.execute(select(Article).where(Article.id == 1))).scalar_one()
await session.refresh(article) # 发一条 SELECT，用数据库当前值覆盖对象
print(article.title) # 一定是此刻数据库里的真值
一句话总结：expire_on_commit=True 是「自动定期对账」（费查询、异步里易炸），False + 手动 refresh() 是「平时信缓存、要新就刷新」（省事、可控）。异步开发选后者。
11. get_db 依赖的雏形
FastAPI 每个请求处理函数都需要一个会话，用完就还。靠什么做到「一个请求一个会话，全自动」？——生成器 + yield：
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession
async def get_db() -> AsyncIterator[AsyncSession]:
"""一个请求一个会话：请求进来造一个，请求结束自动关"""
async with session_factory() as session:
yield session # yield 处把会话交给路由函数；路由跑完后才执行退出
在 FastAPI 里配合 Depends（依赖注入，第 08 章的主角）：
from fastapi import Depends
@app.get("/articles")
async def list_articles(db: AsyncSession = Depends(get_db)):
articles = (await db.execute(select(Article))).scalars().all()
return [a.title for a in articles] # 请求结束，会话自动关闭、连接自动归还
现在只需要记住形状：yield 前是「借」，yield 后是「还」。为什么这么设计、异常时怎么保证归还、怎么统一换成 begin() 模式——第 08 章依赖注入见。
12. 本章速查表
| 要点 | 一句话 |
|---|---|
| 安装 | pip install "sqlalchemy[asyncio]" asyncpg，方括号不能丢 |
| 连接串 | postgresql+asyncpg://user:pwd@host:port/db；换库只改这一行 |
| 引擎 | 进程级单例；echo=True 调试用；用完 dispose() |
| 池参数 | pool_size/max_overflow/pool_pre_ping，服务型数据库专用 |
| 三层关系 | 引擎（柜台）→ 池（充电宝柜子）→ 驱动（线） |
| 裸 SQL | 必须 text() 包裹；参数一律 :占位符 + 字典，杜绝注入 |
| 借连接 | 查询 connect()；写入 begin()（自动提交/回滚） |
| 会话工厂 | async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) |
| 增删查改 | add / execute + scalars().all() / scalar_one() / 改属性 / await delete() |
| 提交三式 | 手动挡 → factory() 管关 → factory.begin() 全包（首选） |
| flush vs commit | flush 保存草稿（自己可见），commit 发布（全局可见） |
| 过期与刷新 | 异步关 expire_on_commit，要最新值用 refresh() |
| get_db 雏形 | async with ... yield session，yield 前借、后还（08 章展开） |
下一章：05 · ORM 建模