SQLAlchemy 2.0：事务与持久化

目标：能够用 SQLAlchemy 2.0 建立 SQLite 表、在显式事务中写入和查询消息，并解释 Engine、Session 与 ORM 对象的职责。

SQLAlchemy 提供两层相关 API：Core 负责 SQL 表达式、Engine、Connection 和事务；ORM 在其上增加映射类、Session、identity map 与对象关系。2.0 风格统一使用 select() 等表达式，Core 与 ORM 的事务语义更一致。

关键对象：

Engine：数据库连接配置和连接池入口，不等于一条长期打开的连接。
Mapped class：Python 类与表结构之间的映射。
Session：一个工作单元，跟踪对象并管理事务；不是全局单例。
commit/rollback：决定事务边界。没有明确边界，异常时很容易留下半完成操作。

安装与完整实验

uv add sqlalchemy

建立 src/textlab/db.py：

from __future__ import annotations

from sqlalchemy import String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class MessageRow(Base):
    __tablename__ = "messages"

    id: Mapped[int] = mapped_column(primary_key=True)
    text: Mapped[str] = mapped_column(String(5000))
    category: Mapped[str] = mapped_column(String(50))
    priority: Mapped[int] = mapped_column(default=0)


def build_engine(url: str = "sqlite+pysqlite:///:memory:"):
    return create_engine(url, echo=False)


def demo() -> None:
    engine = build_engine()
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        with session.begin():
            session.add(MessageRow(text="我要退款", category="refund", priority=2))

        rows = session.scalars(
            select(MessageRow).where(MessageRow.category == "refund")
        ).all()
        print([(row.id, row.text, row.category) for row in rows])


if __name__ == "__main__":
    demo()

运行：

uv run python -m textlab.db

预期输出：

[(1, '我要退款', 'refund')]

session.begin() 正常退出时 commit，异常退出时 rollback。外层 with Session(...) 负责关闭 Session。select(MessageRow) 返回 ORM entity；session.scalars(...) 直接提取第一列中的映射对象。

内存 SQLite 的生命周期

sqlite:///:memory: 数据通常绑定连接生命周期。测试中如果不同线程/连接需要共享同一个内存库，必须理解 pool 和连接配置；FastAPI TestClient 可能让简单“每次新连接”方案看不到已建表。初学实验可以：

使用临时文件 SQLite；或
测试时显式配置 StaticPool 和 check_same_thread=False；
不要把生产配置机械复制为测试配置。

引导实验：验证 rollback

在 session.begin() 中先 add，再抛 RuntimeError，外层捕获。开启新 Session 查询，预期没有记录。这个实验比只读文档更能建立“事务是 all-or-nothing 边界”的模型。

try:
    with Session(engine) as session:
        with session.begin():
            session.add(MessageRow(text="bad", category="demo"))
            raise RuntimeError("force rollback")
except RuntimeError:
    pass

with Session(engine) as session:
    assert session.scalar(select(MessageRow).where(MessageRow.text == "bad")) is None

独立练习

实现 MessageRepository.add() 和 MessageRepository.get()，让 repository 接收 session factory，而不是内部创建全局 Engine。为 add/get 写临时 SQLite 测试。

参考实现轮廓

from collections.abc import Callable


class MessageRepository:
    def __init__(self, session_factory: Callable[[], Session]) -> None:
        self._session_factory = session_factory

    def add(self, *, text: str, category: str, priority: int) -> int:
        with self._session_factory() as session:
            with session.begin():
                row = MessageRow(text=text, category=category, priority=priority)
                session.add(row)
                session.flush()
                message_id = row.id
            return message_id

    def get(self, message_id: int) -> MessageRow | None:
        with self._session_factory() as session:
            return session.get(MessageRow, message_id)

在大型应用里不一定应返回 detached ORM row；可以转换为 dataclass/Pydantic output。本课先聚焦事务和 repository 边界。

常见失败