13 SQLAlchemy ORM —— 异步引擎、会话工厂和请求级 Session，一套基础设施搭好

前置阅读: 12-ORM介绍
关键词: create_async_engine, async_sessionmaker, AsyncSession, aiomysql
难度: ★★★☆☆

场景导入：四件套配齐，才能写出第一个 await db.execute

上一篇从概念层面对比了 ORM 与原生 SQL。但如果你想在 FastAPI 的异步路由里写出 await db.execute(select(Book))，光有概念远远不够——你需要先把 Engine、Session 工厂、模型基类和请求级依赖这四件基础设施搭好。

这四样东西的分工很明确：Engine 管理 aiomysql 连接池，全局只有一个；Session 工厂按请求创建独立的 AsyncSession；模型基类统一注册所有表的元数据；依赖函数以 yield 形态在请求开始时产出 Session、在响应返回后自动关闭。本篇把 app/db.py 写完整，让第 14 章可以直接声明 Author 和 Book 模型。

原理解析：Engine → Session 工厂 → 请求级 Session，逐层看

数据层的边界由三个核心对象组成，异步版本和同步版本只是实现不同，职责完全一致：

（图注：AsyncEngine 通过 AsyncAdaptedQueuePool 管理 aiomysql 连接；工厂按请求创建 AsyncSession；模型统一注册在 DeclarativeBase 的元数据中。）

Engine 是应用级单例。create_async_engine(DATABASE_URL) 接收 mysql+aiomysql://... URL，创建 AsyncEngine 并内置一个 AsyncAdaptedQueuePool——它把连接池的借还操作适配到 asyncio，避免在事件循环里调用同步阻塞池。Engine 只管连接配置和池化，每个进程只创建一次。

async_sessionmaker 是 AsyncSession 的工厂。你把 Engine 传进去，设置 expire_on_commit=False 等默认参数，之后每次调用 AsyncSessionLocal() 就得到一个全新的 AsyncSession。工厂本身不持有连接——连接是在 Session 真正执行查询时才从池里借的。

Base 是声明式模型的注册中心。SQLAlchemy 2.x 推荐显式继承 DeclarativeBase，后续模型使用 Mapped[...] 和 mapped_column 注册到 Base.metadata。

get_async_db 是 yield 依赖。它以 async with AsyncSessionLocal() as session: yield session 形态，在请求开始时创建 Session、在响应返回后自动关闭。框架保证异常路径也走清理逻辑。

下面这张时序图展示从应用启动到请求使用 Session 的完整过程：

（图注：Engine 和 Session 工厂在模块加载时准备就绪，请求到来后才由 AsyncSession 从异步池借连接，请求结束后归还。）

这套设计有三个意图。分层：Engine 只关心连接与池，工厂只关心会话默认值，Base 只关心模型注册——任何一层的修改不影响其他层。生命周期安全：Engine 全局共享以复用连接池，AsyncSession 每请求一个以隔离事务状态。测试便利：测试时可以用 app.dependency_overrides[get_async_db] 把 Session 指向测试库，业务代码零改动。

代码实现：四件套一次配齐

下面的 app/db.py 是整个系列数据层的基石，后续所有章节都依赖它：

# app/db.py
import os
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase

DB_USER = os.getenv("DB_USER", "root")
DB_PASSWORD = os.getenv("DB_PASSWORD", "")
DB_HOST = os.getenv("DB_HOST", "127.0.0.1")
DB_PORT = os.getenv("DB_PORT", "3306")
DB_NAME = os.getenv("DB_NAME", "blog")

DATABASE_URL = (
    f"mysql+aiomysql://{DB_USER}:{DB_PASSWORD}"
    f"@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4"
)

engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    pool_size=10,
    max_overflow=20,
    pool_recycle=1800,
    pool_pre_ping=True,
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    expire_on_commit=False,
    class_=AsyncSession,
)


class Base(DeclarativeBase):
    """SQLAlchemy 2.x 推荐基类，所有模型继承自此。"""


async def init_db() -> None:
    """启动期一次性建表，仅适合开发与小项目。"""
    from app import models  # noqa: F401
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


async def get_async_db() -> AsyncIterator[AsyncSession]:
    """yield 形态的数据库会话依赖，响应返回前自动关闭。"""
    async with AsyncSessionLocal() as session:
        yield session

几个配置项值得单独说明：

• charset=utf8mb4：保证中文、emoji 等完整 Unicode 字符能写入 MySQL。utf8（不带 mb4）只支持基本多语言平面，emoji 插入会报错。

• pool_pre_ping=True：每次从池中取连接时，SQLAlchemy 先发一次 SELECT 1 探测连接是否存活。失效连接会被丢弃并重建。适合云数据库和 NAT 环境下的长连接场景。

• expire_on_commit=False：提交后 ORM 对象的所有属性保持可读，不会因为"过期"而在序列化阶段触发隐式 SELECT。这是 FastAPI 返回 ORM 对象时几乎必开的配置——没开的话，response_model 序列化阶段每个字段都可能触发一次数据库查询。

• pool_size=10, max_overflow=20：常规保持 10 个连接，突发流量时最多再开 20 个。pool_recycle=1800 让连接在 30 分钟后主动回收，防止 MySQL 的 wait_timeout 先于连接池感知而断开。

后续 CRUD 中，所有数据库 I/O 都保持显式等待的形态：

# 查询与写入示意
result = await db.execute(stmt)
await db.commit()
await db.refresh(entity)

避坑指南

• Engine 只创建一个。 它是应用级单例，应在 app/db.py 模块加载时创建并让整个进程共享。在请求函数里反复调用 create_async_engine 会导致每个请求重建连接池，短时间内耗尽 MySQL 的 max_connections。

• AsyncSession 不能跨请求共享。 它不是线程安全的，也不能在多个并发协程间复用。每个请求从 AsyncSessionLocal() 拿到独立的 Session，用完即弃。需要并发时让每个任务创建自己的 Session 作用域。

• expire_on_commit=False 的真正含义。 它让 commit 后对象的属性继续可读——适合 FastAPI 的场景。如果设为 True（默认），事务外访问属性可能触发不可 await 的隐式 I/O，在异步栈下直接报错。

• pool_pre_ping 的代价。 它让每次借连接多一次 SELECT 1 往返。本地稳定环境可以关闭以减少开销，生产环境（尤其是云数据库或 NAT 穿透）通常应该保留。

面试 QA

Q1 [源码]: SQLAlchemy 2.x 中 DeclarativeBase 与 declarative_base() 有什么区别？

DeclarativeBase 是 2.x 推荐的显式基类写法：通过继承定义模型基类，可以自然配合 Mapped[...] 和 mapped_column 的类型注解。declarative_base() 是传统的工厂函数，调用后返回一个声明式基类，在旧项目中仍然可用。

两者都提供 metadata 和模型注册能力，差异主要在声明风格和类型工具支持。本系列使用 DeclarativeBase 配合异步栈，让模型与 AsyncSession 的类型边界更清晰，IDE 自动补全更准确。

Q2 [项目]: Engine、AsyncSessionLocal 与请求依赖的生命周期如何划分？

Engine 是应用级单例，模块加载时创建并持有连接池。AsyncSessionLocal 是绑定 Engine 的工厂，本身不持有连接——每次调用才从池里借。get_async_db 以 async with 管理单个 Session 的完整生命周期：请求进入时创建，响应返回后自动关闭。

多 worker 部署时，每个 worker 进程有自己独立的 Engine，所以 Engine 数量 = worker 数。连接池的容量（pool_size + max_overflow）要按这个因子核算——4 个 worker，每个 pool_size=10，MySQL 最多会收到 40 个连接。

小结

本章把 SQLAlchemy 2.x 异步栈的四件套落地到了 app/db.py：engine 管理 aiomysql 连接池，AsyncSessionLocal 按请求创建会话，Base 统一注册模型元数据，get_async_db 以 yield 依赖形态管理 Session 生命周期。下一篇《14 ORM 建表》将复用 Base 声明 Author 和 Book 模型，用 Mapped 与 mapped_column 把 Python 类型和列约束写在一起，并在启动时完成异步建表。