Python 基础体系 · 第 83/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

SQLAlchemy 2.0：Engine、Session、映射、查询、事务和 N+1

SQLAlchemy 2.0 不是“把 Python 对象自动变成 SQL”的简单工具，而是由两层 API 组成：

• Core：描述 SQL 表达式、连接、事务和结果集；
• ORM：在 Core 之上增加 Python 类与数据库表之间的映射、对象状态管理、身份映射和工作单元。

SQLAlchemy 2.0 的 ORM 查询统一采用 Core 风格的 select()，通过 Session.execute() 或 Session.scalars() 执行；旧式 Session.query() 仍可能存在于兼容代码中，但已经不是 2.0 风格的主 API。(docs.sqlalchemy.org)

本文使用 Python 3.14 语法，示例数据库为 SQLite，SQLAlchemy 版本以 2.0 系列为范围。生产环境应固定 SQLAlchemy、数据库驱动和迁移工具的版本，并在目标数据库上运行测试。

---

一、先建立整体模型：对象是如何走到数据库的

一次典型的 ORM 查询包含下面几层：

flowchart LR
    A[Python 映射类] --> B[select / insert / update]
    B --> C[Session]
    C --> D[Engine]
    D --> E[Dialect]
    D --> F[Connection Pool]
    F --> G[DBAPI Connection]
    G --> H[数据库]
    H --> G
    G --> I[Result]
    I --> C
    C --> J[ORM 对象与 Identity Map]

各组件职责不同：

• 映射类描述“表结构如何对应到 Python 类”；
• **select()**描述要执行的 SQL；
• **Session**管理 ORM 对象、脏数据、刷新、提交和回滚；
• **Engine**提供数据库方言和连接池；
• **Dialect**把通用 SQLAlchemy 表达式编译成目标数据库的 SQL；
• DBAPI 驱动负责真正调用数据库，例如 SQLite、PostgreSQL 或 MySQL 驱动；
• Identity Map 确保同一 Session 中，同一数据库主键通常对应唯一的 Python 对象实例。

Engine 本身并不等于一个数据库连接。它通常持有连接池和方言，并在第一次需要连接时才建立实际 DBAPI 连接；应用进程中通常按数据库 URL 创建并长期复用一个 Engine。(docs.sqlalchemy.org)

---

二、Engine：数据库访问的基础设施

2.1 创建 Engine

最小示例：

from sqlalchemy import create_engine

engine = create_engine(
    "sqlite:///example.db",
    echo=True,
)

这里：

• sqlite:///example.db 表示使用当前目录下的 SQLite 文件；
• echo=True 会输出 SQL 和参数，适合学习和诊断；
• engine 包含 SQLite 方言、连接池和连接创建逻辑；
• 创建 engine 时通常不会马上打开数据库连接，第一次执行数据库操作时才会连接。

生产环境通常关闭 echo，改用结构化 SQL 日志或数据库侧监控，因为 SQL 日志可能包含敏感参数。连接池还会在连接归还时重置事务状态，避免上一个请求留下未提交事务或锁。(docs.sqlalchemy.org)

2.2 Core 直接执行 SQL

即使使用 ORM，也应该理解 Engine 和 Connection 的直接用法：

from sqlalchemy import text

with engine.begin() as connection:
    connection.execute(
        text("CREATE TABLE IF NOT EXISTS message (id INTEGER PRIMARY KEY, body TEXT)")
    )
    connection.execute(
        text("INSERT INTO message (body) VALUES (:body)"),
        {"body": "hello"},
    )

engine.begin() 做了三件事：

1. 从连接池取得连接；
2. 开启事务；
3. 代码块正常结束时提交，发生异常时回滚。

等价的控制流可以写成：

with engine.connect() as connection:
    transaction = connection.begin()
    try:
        connection.execute(
            text("INSERT INTO message (body) VALUES (:body)"),
            {"body": "hello"},
        )
        transaction.commit()
    except Exception:
        transaction.rollback()
        raise

两种写法的差异不在事务语义，而在异常路径是否由上下文管理器统一处理。

2.3 Engine、Connection 和 Session 的边界

可以把它们理解成三个层次：

对象	主要职责	典型使用场景
Engine	管理方言、连接池和连接创建	应用级全局对象
Connection	一次具体数据库连接上的 SQL 与事务	Core、批处理、底层 SQL
Session	ORM 对象、工作单元和事务	Web 请求、业务服务

使用 ORM 时，通常由 Session 间接使用 Engine 和 Connection。SQLAlchemy 官方文档也将 Session 视为 ORM 场景下的事务控制入口，而不是直接操作底层 Transaction。(docs.sqlalchemy.org)

---

三、映射：数据库行如何对应 Python 对象

3.1 Declarative Mapping

SQLAlchemy 2.0 推荐使用类型标注形式的声明式映射：

from __future__ import annotations

from datetime import datetime

from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    fullname: Mapped[str | None] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.current_timestamp()
    )

    addresses: Mapped[list[Address]] = relationship(
        back_populates="user",
        cascade="all, delete-orphan",
    )


class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(200), unique=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))

    user: Mapped[User] = relationship(back_populates="addresses")

这里需要区分几个概念。

Base

class Base(DeclarativeBase):
    pass

Base 是映射类的注册入口。继承它的类会把表结构信息注册到 Base.metadata 中。

__tablename__

__tablename__ = "user_account"

它指定数据库表名。Python 类名可以是 User，表名可以是 user_account，两者不必相同。

Mapped[T]

id: Mapped[int]
fullname: Mapped[str | None]

Mapped[T] 表示这个属性参与 ORM 映射，并且其 Python 类型为 T。

Mapped[str | None] 表示该字段允许为 None。类型标注不仅帮助静态检查，也参与 SQLAlchemy 2.0 对列可空性的推断。为了避免类型和数据库约束不一致，建议显式写出 nullable=False 或让类型标注清晰表达意图。

mapped_column()

name: Mapped[str] = mapped_column(
    String(50),
    unique=True,
    index=True,
)

mapped_column() 描述列的数据库属性：

• String(50)：字符串类型和长度；
• primary_key=True：主键；
• unique=True：唯一约束；
• index=True：创建索引；
• ForeignKey(...)：外键约束；
• server_default=...：由数据库生成默认值。

Python 默认值和数据库默认值不是一回事：

# Python 在 INSERT 前生成
created_at: Mapped[datetime] = mapped_column(
    default=datetime.now,
)

# 数据库在 INSERT 时生成
created_at: Mapped[datetime] = mapped_column(
    server_default=func.current_timestamp(),
)

Python 默认值适合依赖应用进程的值；数据库默认值更接近数据完整性约束，且可以被其他写入程序统一使用。

3.2 relationship 不是外键列

下面两行职责不同：

user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
user: Mapped[User] = relationship(back_populates="addresses")

• user_id 是真实存在于 address 表中的外键列；
• user 是 ORM 在 Python 对象层提供的关系属性；
• relationship() 默认不会创建数据库列；
• back_populates 让两端关系保持一致。

因此：

address = Address(email="a@example.com", user=user)

比手动只设置 user_id 更能表达对象关系。ORM 在 flush 时会根据 user 对象的主键，把 user_id 写入数据库。

3.3 创建表与迁移的边界

学习阶段可以这样创建表：

Base.metadata.create_all(engine)

它会根据当前元数据创建不存在的表，但它不是完整的数据库迁移方案：

• 不会自动安全地管理字段重命名；
• 不会替你设计生产数据迁移；
• 不会记录每次结构变化；
• 不适合多人协作下的版本演进。

生产项目通常使用 Alembic 保存迁移脚本。Alembic 的 --autogenerate 是根据 ORM 元数据和数据库现状生成候选迁移代码，生成结果仍需要人工审查，尤其要检查重命名、数据转换、约束变化和索引变化。(alembic.sqlalchemy.org)

---

四、Session：对象工作单元与事务边界

4.1 Session 不等于连接池

Session 是一个可变、有状态的 ORM 工作单元。它通常包含：

• 当前事务状态；
• 待插入、待更新、待删除对象；
• Identity Map；
• 对数据库连接的临时引用；
• flush、commit、rollback 所需的状态。

Session 通常从 Engine 的连接池中取得连接，但它本身不是连接池，也不应该被多个并发线程或异步任务共享。一个 Session 对应一个非并发使用的事务上下文；并发线程应使用各自的 Session，并发异步任务应使用各自的 AsyncSession。(docs.sqlalchemy.org)

4.2 sessionmaker 与生命周期

from sqlalchemy.orm import sessionmaker

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

SessionLocal 是一个工厂，不是某个具体的 Session。

典型服务函数：

def create_user(name: str, fullname: str | None) -> User:
    with SessionLocal.begin() as session:
        user = User(name=name, fullname=fullname)
        session.add(user)
        session.flush()

        print(user.id)
        return user

生命周期是：

创建 Session
    ↓
添加对象或执行查询
    ↓
Session 自动开始事务
    ↓
flush：把内存变化发送给数据库
    ↓
commit：提交数据库事务
    ↓
Session 退出并释放连接

Session 具有 autobegin 行为：初始状态可能没有事务，但在 add()、execute()、修改持久对象或需要数据库连接时，会自动进入事务状态。事务持续到 commit()、rollback() 或 close()。(docs.sqlalchemy.org)

4.3 Identity Map：为什么两次查询可能得到同一对象

with SessionLocal() as session:
    first = session.scalars(
        select(User).where(User.id == 1)
    ).one()

    second = session.scalars(
        select(User).where(User.id == 1)
    ).one()

    assert first is second

在同一个 Session 中，Identity Map 以“映射类 + 主键”作为对象身份。相同数据库身份通常只维护一个 Python 实例。这样做可以避免：

first.name = "new-name"

之后另一次查询又创建一个代表同一行但包含旧状态的第二个对象。

但 Identity Map 不是全局缓存：

• 不跨 Session 共享；
• 不自动知道其他事务已经提交的变化；
• 受到事务隔离级别和当前数据库连接视图影响；
• 需要显式 refresh()、expire() 或 populate_existing 时才会重新读取。

4.4 commit 后对象为什么可能重新查询

默认情况下，Session.commit() 后会使关联对象的属性过期。之后访问属性时，SQLAlchemy 可能重新发出查询，以便在新事务中获取数据。若对象已经脱离 Session，访问过期属性可能出现：

DetachedInstanceError:
Instance ... is not bound to a Session

如果服务层需要在关闭 Session 后返回已经完整填充的对象，可以使用：

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

但 expire_on_commit=False 不是“数据永远新鲜”。它只是避免提交后自动清空本地属性；长时间复用对象仍可能读取旧值。官方文档明确区分了对象过期、刷新和重新填充三种操作。(docs.sqlalchemy.org)

---

五、flush、commit、rollback：三个容易混淆的动作

5.1 flush：发送 SQL，但不提交事务

with SessionLocal() as session:
    user = User(name="alice")
    session.add(user)

    print(user.id)  # 可能还是 None

    session.flush()

    print(user.id)  # 数据库生成主键后通常可用
    session.rollback()

flush() 的作用是把 Session 中的对象变化同步到当前数据库事务：

Python 对象变化
    ↓ flush
INSERT / UPDATE / DELETE
    ↓
当前事务中的数据库状态

flush() 之后仍然可以回滚。它不是提交。

SQLAlchemy ORM 会在某些需要保证查询结果正确的时机自动 flush，例如执行查询或提交前。因为如果刚刚修改了对象却直接查询，查询可能需要看到这些修改。官方文档称之为 autoflush。(docs.sqlalchemy.org)

5.2 commit：提交事务

with SessionLocal() as session:
    user = User(name="bob")
    session.add(user)
    session.commit()

commit() 通常会先 flush，再提交数据库事务：

Session 中的变化
    ↓ flush
数据库事务内的变化
    ↓ commit
对其他事务可见，且不能用 rollback 撤销

如果只是执行查询而没有写入，通常不需要调用 commit()。(docs.sqlalchemy.org)

5.3 rollback：撤销当前事务和未提交的对象变化

with SessionLocal() as session:
    try:
        session.add(User(name="duplicate"))
        session.flush()
        session.commit()
    except Exception:
        session.rollback()
        raise

如果 flush() 因唯一约束失败，Session 会进入需要回滚的失败状态。此时不能继续正常执行其他 SQL，必须先：

session.rollback()

一个常见错误是：

try:
    session.commit()
except Exception:
    print("写入失败")

session.execute(select(User))  # 可能继续报错

正确做法是回滚后再决定重试、返回业务错误或抛出异常。

5.4 嵌套事务与 SAVEPOINT

begin_nested() 创建的是数据库 SAVEPOINT，而不是独立的顶层事务：

with SessionLocal.begin() as session:
    session.add(User(name="outer"))

    try:
        with session.begin_nested():
            session.add(User(name="maybe-invalid"))
            session.flush()
    except Exception:
        # 回滚到 SAVEPOINT，外层事务仍可继续
        pass

    session.add(User(name="outer-continued"))

状态变化大致为：

外层事务 BEGIN
    ├── 写入 outer
    ├── SAVEPOINT
    │   ├── 写入 maybe-invalid
    │   └── 失败 → ROLLBACK TO SAVEPOINT
    ├── 写入 outer-continued
    └── COMMIT

这适合批量导入中“单条失败但整体继续”的场景。不过 SAVEPOINT 仍受数据库驱动和数据库后端支持影响，不能把它当成跨数据库完全一致的行为。

---

六、查询：2.0 风格的 select、Result 和 scalars

6.1 基本查询

from sqlalchemy import select

with SessionLocal() as session:
    statement = (
        select(User)
        .where(User.name == "alice")
        .order_by(User.id)
    )

    result = session.execute(statement)
    users = result.scalars().all()

这里的结果转换分为两步：

result = session.execute(statement)
users = result.scalars().all()

• execute() 返回 Result；
• select(User) 的每一行默认是一个包含 ORM 对象的行；
• scalars() 取出每行的第一个元素；
• all() 把结果收集为列表。

也可以直接写：

with SessionLocal() as session:
    users = session.scalars(
        select(User).where(User.name.like("a%"))
    ).all()

6.2 one、first、scalar_one_or_none

不同方法表达不同的数据库结果约束：

user = session.scalars(
    select(User).where(User.id == user_id)
).one()

• one()：必须恰好一行；零行或多行都会异常；
• first()：取第一行，零行返回 None，不会表达“必须唯一”；
• one_or_none()：零行返回 None，多行异常；
• scalar_one_or_none()：当查询返回标量列时使用。

按主键查询时，也可以使用：

user = session.get(User, user_id)

get() 具有 Identity Map 语义：如果当前 Session 已经有这个主键对应的对象，可能直接返回该对象；否则才查询数据库。

6.3 查询多个实体

statement = (
    select(User, Address)
    .join(User.addresses)
    .where(Address.email.like("%@example.com"))
)

rows = session.execute(statement).all()

for user, address in rows:
    print(user.name, address.email)

因为 select(User, Address) 选择了两个实体，所以每一行包含两个元素。不能盲目调用 scalars() 后期待同时得到两个对象；scalars() 只取每行的第一个元素。

6.4 JOIN 与 ORM 对象去重

集合关系使用 joinedload() 时，一个父对象可能因多个子对象而在 SQL 结果中出现多次：

from sqlalchemy.orm import joinedload

statement = select(User).options(joinedload(User.addresses))
users = session.scalars(statement).unique().all()

unique() 的作用是对 ORM 实体结果进行去重。其原因不是 Python 列表本身，而是 SQL JOIN 的行数变化：

User 1 + Address A → 一行
User 1 + Address B → 一行
User 2 + Address C → 一行

数据库返回三行，但 ORM 中可能只希望得到两个 User 对象。SQLAlchemy 对这种 joined eager loading 结果要求显式 unique()，防止调用者无意中忽略 JOIN 造成的基数变化。

---

七、写入：添加、修改和删除

7.1 添加对象

with SessionLocal.begin() as session:
    user = User(
        name="carol",
        fullname="Carol Zhang",
        addresses=[
            Address(email="carol@example.com"),
            Address(email="carol.work@example.com"),
        ],
    )
    session.add(user)

由于 Address.user 与 User.addresses 已通过关系配置，flush 时 ORM 可以推导出插入顺序：

先 INSERT user_account
    ↓ 获得 user.id
再 INSERT address，并填入 user_id

这里的 cascade="all, delete-orphan" 表示通过父对象关系管理子对象：

• 把 Address 放进 user.addresses 时，子对象会参与持久化；
• 从集合中移除并且没有其他父对象引用时，可能被删除；
• 删除用户时，相关地址也会按关系级联处理。

级联规则必须结合数据库外键、ON DELETE 和业务语义验证，不能仅凭参数名称推断最终行为。

7.2 修改持久化对象

with SessionLocal.begin() as session:
    user = session.get(User, 1)
    if user is None:
        raise LookupError("user not found")

    user.fullname = "New Name"

ORM 会通过对象状态跟踪发现 fullname 发生变化，在 flush 时生成类似：

UPDATE user_account
SET fullname = ?
WHERE user_account.id = ?

这依赖的是 ORM 的单位工作机制，而不是 Python 属性赋值本身直接执行 SQL。

7.3 删除对象

with SessionLocal.begin() as session:
    user = session.get(User, 1)
    if user is not None:
        session.delete(user)

是否同时删除地址，取决于 ORM 级联配置和数据库外键策略。删除父对象前，应明确：

1. 子记录是否必须删除；
2. 子记录是否允许保留；
3. 外键是否设置 ON DELETE CASCADE；
4. ORM 是否需要 passive_deletes=True；
5. 数据库是否真的启用了外键约束。

---

八、N+1：不是“查询多”这么简单

8.1 N+1 的形式化定义

假设先查询出 N 个用户：

users = session.scalars(select(User)).all()

随后在循环中访问每个用户的地址：

for user in users:
    print(user.name, user.addresses)

如果 addresses 使用默认的懒加载，执行过程可能是：

第 1 条 SQL：查询 N 个 User
第 2 条 SQL：查询 User 1 的 addresses
第 3 条 SQL：查询 User 2 的 addresses
...
第 N+1 条 SQL：查询 User N 的 addresses

总查询次数为：

$$Q(N) = 1 + N$$

其中：

• 1 是父对象查询；
• N 是每个父对象一次关系查询。

这就是 N+1。

问题不只是 SQL 数量，还包括：

• 每次查询都有网络往返；
• 数据库需要重复解析和执行相似语句；
• 延迟大致受 N 次往返叠加影响；
• 查询发生在属性访问处，代码表面上没有写 SQL；
• Session 关闭后访问懒加载属性会失败；
• 异步场景中的隐式 IO 可能无法在普通属性访问中执行。

SQLAlchemy 官方文档也特别指出，懒加载可能产生额外查询，且这些查询是隐式发出的；在对象脱离 Session 或使用某些异步并发模式时还可能直接失败。(docs.sqlalchemy.org)

8.2 可复现的 N+1 示例

from sqlalchemy import event, select

query_count = 0


@event.listens_for(engine, "before_cursor_execute")
def count_queries(
    conn,
    cursor,
    statement,
    parameters,
    context,
    executemany,
):
    global query_count
    query_count += 1


with SessionLocal.begin() as session:
    users = session.scalars(select(User)).all()

    for user in users:
        _ = user.addresses

print(query_count)

如果数据库中有 N 个用户，且每个关系都未预加载，查询次数可能接近 N + 1。需要注意，实际次数还会受关系是否已经加载、Identity Map 是否已有对象、关系类型以及数据库方言影响。

8.3 selectinload：通常适合集合关系

from sqlalchemy.orm import selectinload

statement = (
    select(User)
    .options(selectinload(User.addresses))
)

with SessionLocal() as session:
    users = session.scalars(statement).all()

    for user in users:
        print(user.name, user.addresses)

典型 SQL 形态是：

-- 查询用户
SELECT ...
FROM user_account;

-- 使用 IN 一次查询这些用户的地址
SELECT ...
FROM address
WHERE address.user_id IN (?, ?, ?, ...);

查询次数通常从：

$$1 + N$$

降为：

$$1 + B$$

其中 B 是 IN 批次数量。用户数量不超过一次 IN 可容纳的主键数量时，通常就是两次查询。

selectinload 不会把子表行直接乘到父查询结果中，因此对一对多集合关系通常比 joinedload 更容易控制结果行数。它仍然可能分批发出多个查询，不能简单说“永远只有两条 SQL”。

8.4 joinedload：一次 JOIN，但可能放大结果集

from sqlalchemy.orm import joinedload

statement = select(User).options(
    joinedload(User.addresses)
)

with SessionLocal() as session:
    users = session.scalars(statement).unique().all()

典型 SQL 形态：

SELECT user_account.*, address.*
FROM user_account
LEFT OUTER JOIN address
    ON user_account.id = address.user_id;

如果：

• 用户数为 U；
• 每个用户平均有 A 个地址；

那么 JOIN 结果行数大约为：

$$R \approx U \times A$$

当 A 很大或同时 JOIN 多个集合关系时，结果集可能发生乘法膨胀。例如：

User
 ├── addresses：10 条
 └── orders：20 条

同时 JOIN 两个集合，单个用户可能得到：

$$10 \times 20 = 200$$

行，尽管最终只需要一个 User 对象。因此 joinedload 的查询次数少，不代表传输数据量和数据库工作量一定少。

8.5 subqueryload 和 raiseload

subqueryload 通过额外子查询加载关系，历史上用于某些复杂查询，但现代代码通常优先比较 selectinload 和 joinedload 的实际 SQL。

如果某个接口绝不能隐式加载关系，可以使用 raiseload：

from sqlalchemy.orm import raiseload

statement = select(User).options(
    raiseload(User.addresses)
)

with SessionLocal() as session:
    user = session.scalars(statement).first()
    if user is not None:
        user.addresses  # 直接抛出异常，而不是偷偷执行 SELECT

这适合测试和边界明确的读模型，因为它把“潜在 N+1”变成了显式失败。关系加载策略包括默认懒加载、joined、selectin、subquery 等，SQLAlchemy 将这些策略作为查询级 loader options 配置。(docs.sqlalchemy.org)

---

九、关系加载策略的选择逻辑

可以按“关系基数”和“返回用途”推导，而不是机械套用某一个选项。

一对多集合

例如一个用户有多个地址：

select(User).options(selectinload(User.addresses))

通常先考虑 selectinload，因为：

• 父查询结果不会因子表数量直接膨胀；
• 子查询可以使用 WHERE child.parent_id IN (...)；
• 适合批量读取父对象及其集合。

多对一引用

例如地址属于一个用户：

select(Address).options(joinedload(Address.user))

多对一通常每条子记录只对应一个父记录，JOIN 的结果放大风险较低，joinedload 可能更直接。

只需要聚合结果

如果接口只需要地址数量，不需要完整地址对象，不应加载整个关系：

from sqlalchemy import func

statement = (
    select(User.id, User.name, func.count(Address.id))
    .outerjoin(User.addresses)
    .group_by(User.id, User.name)
)

rows = session.execute(statement).all()

这是“改变查询目标”而不是“优化对象加载”。相比先加载用户、再加载地址、最后在 Python 中计数，它直接让数据库执行聚合。

分页场景

对父表分页时，selectinload 通常更容易理解：

statement = (
    select(User)
    .order_by(User.id)
    .limit(20)
    .offset(40)
    .options(selectinload(User.addresses))
)

先确定当前页的用户，再按当前页主键加载地址，避免 JOIN 后分页把“行”误当成“用户”分页。

---

十、完整可运行示例

下面示例包含建表、写入、查询、预加载和事务回滚。

from __future__ import annotations

from sqlalchemy import ForeignKey, String, create_engine, select
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    mapped_column,
    relationship,
    selectinload,
)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    addresses: Mapped[list[Address]] = relationship(
        back_populates="user",
        cascade="all, delete-orphan",
    )


class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(200), unique=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped[User] = relationship(back_populates="addresses")


engine = create_engine("sqlite:///demo.db")
Base.metadata.create_all(engine)


def seed() -> None:
    with Session(engine) as session:
        alice = User(
            name="alice",
            addresses=[
                Address(email="alice@example.com"),
                Address(email="alice.work@example.com"),
            ],
        )
        bob = User(
            name="bob",
            addresses=[
                Address(email="bob@example.com"),
            ],
        )

        session.add_all([alice, bob])
        session.commit()


def list_users() -> None:
    statement = (
        select(User)
        .order_by(User.id)
        .options(selectinload(User.addresses))
    )

    with Session(engine) as session:
        users = session.scalars(statement).all()

        for user in users:
            print(
                user.id,
                user.name,
                [address.email for address in user.addresses],
            )


def demonstrate_rollback() -> None:
    with Session(engine) as session:
        try:
            session.add(User(name="alice"))
            session.commit()
        except Exception as exc:
            session.rollback()
            print(type(exc).__name__)


if __name__ == "__main__":
    seed()
    list_users()
    demonstrate_rollback()

预期输出类似：

1 alice ['alice@example.com', 'alice.work@example.com']
2 bob ['bob@example.com']
IntegrityError

执行 seed() 的前置条件是当前目录可写，并且 demo.db 不包含相同唯一值。重复运行时，alice 和 bob 的唯一约束会触发 IntegrityError，这正好展示了为什么异常后必须调用 rollback()。

这个示例中的关键点是：

1. Base.metadata.create_all(engine) 只负责学习阶段的建表；
2. Session(engine) 创建一次工作单元；
3. session.add_all() 只把对象加入 Session；
4. commit() 触发 flush 并提交事务；
5. selectinload() 在访问 user.addresses 前主动加载集合；
6. rollback() 将失败事务恢复到可继续使用的状态。

---

十一、Web 服务中的 Session 生命周期

在 Web 应用中，通常应让一次请求使用一个 Session：

请求开始
    ↓
创建 Session
    ↓
执行查询或修改
    ↓
成功：commit
失败：rollback
    ↓
请求结束：close

伪代码如下：

def handle_request() -> dict:
    with SessionLocal() as session:
        try:
            user = session.scalars(
                select(User).where(User.name == "alice")
            ).one_or_none()

            if user is None:
                return {"error": "not found"}

            user.fullname = "Alice Updated"
            session.commit()

            return {
                "id": user.id,
                "name": user.name,
            }
        except Exception:
            session.rollback()
            raise

更推荐将“事务边界”放在服务层，而不是让每一个底层 Repository 方法都随意提交：

def update_user_name(user_id: int, name: str) -> None:
    with SessionLocal.begin() as session:
        user = session.get(User, user_id)
        if user is None:
            raise LookupError("user not found")

        user.name = name

如果一个业务操作包含：

创建订单
扣减库存
写入审计日志

这三个动作需要在同一个事务中成功或失败，就必须共享同一个 Session，而不是让三个函数各自创建 Session 并分别提交。否则可能出现：

创建订单成功并提交
扣减库存失败
审计日志没有写入

此时已经无法通过最后一个函数的 rollback 撤销前面的提交。

---

十二、并发：为什么不能共享 Session

下面的模式是错误的：

shared_session = SessionLocal()

def worker():
    shared_session.execute(select(User))

如果多个线程或异步任务同时调用它，会让同一个可变对象同时处理多个数据库操作。Session 还要维护当前事务、flush 状态、对象状态和连接状态，因此并发修改会产生非法状态。

正确模型是：

def worker():
    with SessionLocal() as session:
        return session.scalars(select(User)).all()

异步版本则是每个任务创建自己的 AsyncSession。SQLAlchemy 将这一原则概括为：

Session per thread
AsyncSession per task

单个数据库事务本身也不是为了同时接收多个并发 SQL 命令而设计的；如果应用需要并发数据库操作，应使用多个并发事务，而不是让多个任务共享一个事务。(docs.sqlalchemy.org)

---

十三、异步数据库访问与隐式懒加载

SQLAlchemy 的异步 ORM 仍然使用 AsyncSession，但异步不改变事务和 Session 的核心约束：

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy import select
from sqlalchemy.orm import selectinload

async_engine = create_async_engine(
    "sqlite+aiosqlite:///async-demo.db"
)

AsyncSessionLocal = async_sessionmaker(
    async_engine,
    expire_on_commit=False,
)


async def get_users() -> list[User]:
    async with AsyncSessionLocal() as session:
        statement = select(User).options(
            selectinload(User.addresses)
        )
        result = await session.scalars(statement)
        return result.all()

异步场景尤其不应依赖普通属性访问触发隐式 IO：

user.addresses

在同步 ORM 中，这个访问可能隐式执行 SELECT；在异步代码中，普通属性访问没有 await 位置，可能导致无法执行或出现异步上下文错误。因此应在查询阶段使用 selectinload()、joinedload()，或显式执行异步加载。

异步数据库访问的完整设计还需要考虑连接池大小、事务取消、任务超时和取消后的 rollback。AsyncSession 同样不是并发安全对象，不能被多个 asyncio 任务共享。(docs.sqlalchemy.org)

---

十四、常见误解与失败表现

误解一：flush() 就是提交

反例：

with SessionLocal() as session:
    session.add(User(name="temporary"))
    session.flush()
    session.rollback()

回滚后，temporary 不会作为已提交数据保留。flush() 只是把 SQL 发送到当前事务。

误解二：查询到对象就代表数据库最新

user = session.get(User, 1)
# 另一个事务修改并提交了同一行
user_again = session.get(User, 1)

user_again 可能仍是 Identity Map 中的原对象。若需要显式刷新：

session.refresh(user)

或者：

user = session.scalars(
    select(User)
    .where(User.id == 1)
    .execution_options(populate_existing=True)
).one()

SQLAlchemy 的默认设计假设当前事务具有较强隔离性；需要读取外部变化时，应明确要求刷新，而不是期待每次查询都覆盖已有对象。(docs.sqlalchemy.org)

误解三：返回 ORM 对象后，Session 关闭也没问题

def get_user() -> User:
    with SessionLocal() as session:
        return session.get(User, 1)

如果调用方随后访问尚未加载的关系：

user.addresses

就可能因为对象已经脱离 Session 而失败。解决办法不是到处捕获异常，而是在 Session 关闭前确定读取边界：

def get_user() -> User:
    with SessionLocal() as session:
        return session.scalars(
            select(User)
            .where(User.id == 1)
            .options(selectinload(User.addresses))
        ).one()

或者不要把 ORM 实体直接作为跨层返回值，而是在事务内转换为 DTO、字典或响应模型。

误解四：使用 joinedload 就没有重复对象问题

集合 JOIN 会产生重复父行：

statement = select(User).options(
    joinedload(User.addresses)
)

users = session.scalars(statement).all()

这类查询应使用：

users = session.scalars(statement).unique().all()

否则结果处理阶段可能抛出要求调用 unique() 的异常，或者产生不符合预期的实体结果。

误解五：把 expire_on_commit=False 当成缓存策略

它只改变提交后的属性过期行为，不提供跨请求缓存、失效通知或并发一致性。长时间保存 ORM 对象会扩大脏数据窗口，也会让对象状态和数据库状态逐渐脱节。

---

十五、诊断 N+1 的方法

15.1 先观察实际 SQL

开发环境可以暂时开启：

engine = create_engine(
    "sqlite:///demo.db",
    echo=True,
)

观察以下模式：

SELECT user_account ...
SELECT address ... WHERE ? = address.user_id
SELECT address ... WHERE ? = address.user_id
SELECT address ... WHERE ? = address.user_id

如果后面重复出现结构相同、仅参数不同的查询，通常就是关系懒加载造成的 N+1。

15.2 在测试中统计查询次数

from contextlib import contextmanager
from sqlalchemy import event


@contextmanager
def count_sql(engine):
    counter = {"value": 0}

    def before_execute(
        conn,
        cursor,
        statement,
        parameters,
        context,
        executemany,
    ):
        counter["value"] += 1

    event.listen(engine, "before_cursor_execute", before_execute)

    try:
        yield counter
    finally:
        event.remove(engine, "before_cursor_execute", before_execute)

测试：

def test_user_list_does_not_have_n_plus_one():
    with count_sql(engine) as counter:
        with SessionLocal() as session:
            users = session.scalars(
                select(User).options(selectinload(User.addresses))
            ).all()

            for user in users:
                len(user.addresses)

    assert counter["value"] <= 2

断言不应简单地固定为“必须两条 SQL”，因为数据库方言、批次大小、额外过滤条件和其他 loader option 都可能改变次数。更可靠的测试是：

• 数据量从 1 增加到 N 时，查询次数不应线性增长；
• 关键接口的 SQL 数量应有明确上限；
• 关系访问应在查询语句中显式声明。

15.3 使用 raiseload 发现隐式查询

statement = select(User).options(
    raiseload("*")
)

这会让未声明的关系访问直接失败。它适合在特定读路径中强制约束加载边界，也适合发现序列化器、模板和循环中隐藏的关系访问。

---

十六、事务一致性与数据库约束

ORM 不能替代数据库约束。下面的代码存在竞态：

with SessionLocal.begin() as session:
    exists = session.scalar(
        select(User.id).where(User.name == "alice")
    )

    if exists is None:
        session.add(User(name="alice"))

两个并发事务可能同时观察到“不存在”，然后都尝试插入。正确的最终防线仍然是数据库唯一约束：

name: Mapped[str] = mapped_column(
    String(50),
    unique=True,
    nullable=False,
)

业务代码需要处理唯一约束异常：

from sqlalchemy.exc import IntegrityError

with SessionLocal() as session:
    try:
        session.add(User(name="alice"))
        session.commit()
    except IntegrityError:
        session.rollback()
        raise ValueError("用户名已存在")

检查再插入可以改善用户体验，但不能代替唯一约束。因为：

SELECT 检查
    ↓
另一个事务同时插入
    ↓
当前事务 INSERT

这段时间窗口就是竞态条件。约束负责保证最终不违反规则，应用负责把数据库异常转换成业务可理解的错误。

---

十七、与 Alembic 的连接

映射代码描述的是“当前期望的数据库结构”，Alembic revision 描述的是“结构如何从旧版本变成新版本”。

例如向 User 增加字段：

nickname: Mapped[str | None] = mapped_column(String(50))

仅修改 Python 类不会自动修改线上数据库。需要：

修改映射类
    ↓
alembic revision --autogenerate
    ↓
人工审查 revision
    ↓
测试数据库执行 upgrade
    ↓
生产环境 upgrade

发布风险主要出现在以下情况：

• 直接删除字段导致旧代码无法运行；
• 新字段设为非空但已有数据无法填充；
• 自动生成把重命名识别成“删除旧字段 + 新增字段”；
• 多个开发分支生成多个 head；
• 数据迁移和代码发布顺序不兼容。

因此，映射、事务和迁移应分开理解：

• 映射解决 Python 与当前数据库结构如何对应；
• Session解决一次业务操作中的对象状态和事务；
• Alembic解决多个数据库结构版本之间如何演进。

Alembic 的自动生成会比较元数据和数据库状态并生成候选 Python 迁移脚本，但候选脚本必须人工检查；迁移脚本还可以形成分支、合并 revision，并通过 upgrade 或 downgrade 进行版本移动。(alembic.sqlalchemy.org)

---

十八、把标题中的概念串成一个可执行判断

面对一段 SQLAlchemy 2.0 代码，可以按以下因果链检查：

Engine 是否是进程级复用？
    ↓
Session 是否只属于一个请求、线程或任务？
    ↓
映射是否正确表达主键、外键、可空性和关系？
    ↓
查询是否使用 select() 并正确处理 Result？
    ↓
写入是否明确 flush、commit、rollback 的边界？
    ↓
关系属性是否会触发隐式懒加载？
    ↓
加载策略是否与返回数据量和分页方式匹配？
    ↓
数据库约束是否承担了并发下的最终一致性？

最小原则不是“所有查询都 eager load”，而是：

1. Engine 长生命周期复用；
2. Session 短生命周期、非并发共享；
3. 事务边界覆盖完整业务操作；
4. 查询结果语义由 one()、first()、scalars() 等方法明确表达；
5. 关系加载策略显式写在查询中；
6. 通过 SQL 日志和查询计数验证是否存在 N+1；
7. 把数据库约束和迁移脚本当作数据一致性的组成部分。

当这些边界清晰后，SQLAlchemy 的对象模型、SQL 表达式和数据库事务并不是三套互相冲突的系统，而是同一条数据流上的不同抽象层。

---

系列导航与关联阅读

• 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
• 上一篇：Python Web API 工程：契约、错误、分页、幂等、限流和版本
• 下一篇：Alembic 数据库迁移：Revision、自动生成、分支、上线和回滚
• 延伸：Python 异步数据库访问：连接池、事务、取消、并发和一致性

官方资料

• Django Documentation
• Django REST Framework Documentation

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。