02 课堂课件 · D02

01课前预习02课堂课件03学生任务04完成检查05课后复习

数据库服务开发：分层、Session 与事务回滚

课前先保留判断：课前预习；课后用你的结果复验：课后复习。

今天从哪里开始

昨天能复核商品文件并分别写入、查询表，今天除了归还文件和数据库资源，还要完成一笔两种商品的订单：第一条已经成功，第二条库存不足。若每一步各自保存，客服会看到一张只有半套明细的订单。

先推演这个坏结果，再明确哪些动作必须一起成功。接着把事务接入业务服务，用 SQLAlchemy 管理对象和连接，最后在中途制造失败，检查订单、明细和库存是否都恢复。

今天在图中一次订单操作的前后补上资源边界：进入取得 Session，退出提交或回滚并关闭。

今天要学会什么

• Service 与持久化边界
• SQLAlchemy Session
• flush、commit、rollback
• Unit of Work 与失败注入

课堂结束时，用自己的项目检查：

• 能做出来：解释 with 的进入/退出与异常传播，第二条明细失败时验证连接收口、所有表和库存恢复。
• 能查出问题：针对“循环内提交、全局 Session、Fake 冒充 MySQL”中的风险，说明输入、预期与实际差异；失败后核对数据没有被误改。
• 能解释修改：提交代码、测试和修改前后结果，讲清自己接受或拒绝了哪一处 AI 建议。

先演示一个坏结果

伪代码：

create_order_header()
commit()
 
reserve_product_a()
commit()
 
reserve_product_b()  # 这里库存不足
commit()

最终可能留下：

• 订单头已存在；
• 第一件库存已预留；
• 第二件没预留；
• 订单无法完整履约。

这就是“半成功”。

---

这笔订单应该保留一半，还是作为一次完整业务处理？

本课下单合同要求订单头、明细和库存一起成功或一起回滚；部分订单需要另行设计业务合同，不能由异常碰巧决定。

事务是什么

事务把一组相关操作当成一个工作单元。

基本动作：

START TRANSACTION;
-- 多条相关 SQL
COMMIT;

发生异常：

ROLLBACK;

COMMIT（提交记录） 之后才对这次事务做最终确认。

ROLLBACK（回滚） 撤销本事务尚未提交的变化。

---

四层职责和依赖方向

下单涉及输入、校验、计算和保存。先确定谁负责组织整个动作，再选择数据库调用方式；否则每个存取函数自己提交，服务层就无法统一撤回。

CLI → Service → Storage
        ↓          ↓
       Model ← 恢复模型

层	负责	不负责
Model	字段、类型、对象不变量	打印、命令参数、文件路径
Storage	保存/读取、Schema（数据结构约束）、损坏诊断	决定能否支付/取消
Service	创建、查询、状态动作和跨对象规则	解析 argparse、直接打印
CLI	参数、调用、输出、退出码	直接改状态、直接读写 JSON

现场审查 AI 方案

AI（人工智能） 建议在 CLI 中执行：

order.order_status = "CANCELLED"
storage.save(dataset)

问题不是“代码不能运行”，而是另一个入口可以绕过取消规则。要求 AI 把动作收回 Service，再用同一 Service 测试保护。

接口、业务服务和数据访问层各负责什么，事务边界应由谁决定？

接口处理输入输出，业务服务组织一次用例，数据访问层执行存取；由用例统一决定提交或回滚，避免每个数据访问动作自行提交。

从文件错误迁移到数据库业务错误

上一周本人订单交付模块的 SnapshotError 用于快照恢复失败；旧文件版本中还存在 StorageCorruptedError，不能混用名称。现在沿用“上层按类型处理”的原则，但使用数据库服务的实际异常：ResourceNotFound、ProductInactive、InsufficientInventory、InvalidOrderTransition、DataConflict。不能把旧版本带 Error 后缀的类名直接导入新项目。

这些异常继承 CommerceError，Service 抛业务错误，明天的 HTTP 适配层再翻译状态码。今天直接调用 Python 服务并捕获异常，核对数据库状态；底层连接错误不能伪装成不存在的订单。

SQLAlchemy：先认识表映射与连接工厂

ORM 把表记录和 Python 对象连接起来，不会替我们决定事务边界，也不消除 SQL 成本。Engine 管理数据库连接池；Session 管理一次工作的状态；映射类说明哪张表的哪些字段对应对象属性。

下面是映射机制的独立结构例，省略数据库连接；本项目的完整字段以已建立的 products 表为准，不执行 create_all 去悄悄修改现有表。

from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
class Base(DeclarativeBase):
    pass
 
class ProductRow(Base):
    __tablename__ = "products"
    product_id: Mapped[str] = mapped_column(String(5), primary_key=True)
    name: Mapped[str] = mapped_column(String(60))

Mapped[str] 表示这个 ORM 属性按字符串映射，mapped_column 描述数据库列；它不等于运行时业务输入校验。接下来再看依赖已有 MySQL URL 和映射类的查询片段：

from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker
 
engine = create_engine(database_url, pool_pre_ping=True)  # 从私有环境配置读取
SessionFactory = sessionmaker(bind=engine)
with SessionFactory() as session:
    row = session.scalar(select(ProductRow).where(ProductRow.product_id == "P1001"))
    print(None if row is None else row.name)

这段是项目片段，不能在缺少 database_url 时直接运行。先让 AI 解释最终 WHERE 条件，再核对日志中的参数化 SQL。不要把密码、连接 URL 或真实客户数据粘进 Prompt。

原子下单只有一个提交点

看图时找唯一的 COMMIT。

它位于全部步骤成功之后：

1. 校验 Customer 与 Product；
2. 锁定需要更新的库存行；
3. 计算服务端金额；
4. 写 Order；
5. 写所有 OrderItem；
6. 更新所有 Inventory；
7. 最后 COMMIT。

循环内部不能逐条 commit。

---

为什么每条订单明细保存后不能立即 commit？

一笔下单要同时保持订单头、明细和库存正确。中间提交会使第二条失败时无法撤回第一条；今天以一个用例一个事务，在全部成功后提交。

autocommit 的陷阱

MySQL 常见默认 autocommit=1。

这意味着一条独立语句成功后可能立即提交。

对多步业务操作，我们必须显式建立事务边界。

不要只看到代码里没有 commit()，就假设一定安全。

Driver 和框架的事务模式需要明确。

今天以受控 Session 边界为准。

---

用失败注入证明 rollback

订单包含两件商品：

• P1001：库存足够；
• P1002：故意设置库存不足，或在第二条明细写入时抛出受控异常。

失败前查询：

SELECT * FROM orders WHERE order_id = :new_order_id;
SELECT * FROM order_items WHERE order_id = :new_order_id;
SELECT product_id, available_qty, reserved_qty
FROM inventory
WHERE product_id IN ('P1001', 'P1002');

失败后执行同样查询。

---

rollback 通过要看到什么

失败后：

• 新 Order 不存在；
• 新 OrderItem 一条也不存在；
• P1001 库存回到操作前；
• P1002 库存保持操作前；
• Python 服务抛稳定业务异常（下次课再接 HTTP）；
• 底层 SQL 错误不泄漏给客户端。

只看到异常被捕获不够。

必须查数据库前后状态。

---

怎样证明失败后没有留下残单，而不只是屏幕提示了失败？

故意在中途注入失败，再查询订单头、明细和库存，比较失败前后值。提示文字不能代替数据库状态检查。

Session 是什么

事务边界已经确定，现在看 Python 怎样把这次工作的对象和 SQL 组织起来。SQLAlchemy 的 Session（工作会话）跟踪这一次操作，不是浏览器登录会话，也不是全局共享数据容器。

Session 是一次数据库工作的状态化工作区。

它会：

• 取得连接；
• 发出 SQL；
• 跟踪对象变化；
• flush 待写操作；
• commit 或 rollback；
• 结束后归还连接。

Session 实例不能全局共享给所有并发请求。

Engine（执行引擎） 和 sessionmaker 可以作为工厂长期存在。

---

为什么 Session 不应作为全局共享对象？

Session 保存当前事务和对象状态；并发请求共用会混淆生命周期和回滚边界。连接池可复用底层连接，每个用例仍需自己的 Session，并及时关闭。

flush、commit、rollback

flush：

• 把待处理的 ORM 变化发成 SQL；
• 事务仍未最终提交；
• 可能得到数据库生成的主键或约束错误。

commit：

• 先 flush；
• 再提交当前事务。

rollback：

• 撤销当前事务未提交变化；
• flush 失败后，同一个 Session 继续使用前要 rollback。

---

flush 与 commit 有什么区别？

flush 把待执行 SQL 发给数据库以取得主键或提前暴露约束错误，但事务仍未最终提交；commit 完成事务，rollback 只能撤回尚未提交的修改。

让连接在成功和失败时都有明确归宿

为什么学：商城下单第二条明细失败时，既要撤回前面的修改，也要归还数据库连接。只写 except: print(...)，可能留下半张订单、占住连接，还让调用者误以为成功。

我昨天用 with 关闭文件，今天需要看清它怎样工作。实现 __enter__ 和 __exit__ 的对象可用作上下文管理器：进入调用 __enter__，其返回值交给 as；退出调用 __exit__(异常类型, 异常对象, traceback)。正常退出时三个参数为 None。

下面是独立机制例，只打印事件，不连接数据库：

class TraceScope:
    def __enter__(self):
        print("进入")
        return self
 
    def __exit__(self, exc_type, exc, tb):
        print("退出", None if exc_type is None else exc_type.__name__)
        return False
 
try:
    with TraceScope() as scope:
        raise ValueError("第二条明细失败")
except ValueError:
    print("调用者仍能看到失败")

输出依次为进入、退出 ValueError、调用者看到失败。return False 或 None 不压住异常；若改成 True，异常会被压住。下单不能因为资源清理完成就谎报成功。

我把异常改成正常打印，再比较输出。若 __enter__ 自身失败，不能依靠该对象的 __exit__ 清理尚未成功进入的资源；获取资源阶段同样要有清楚的责任。

finally 管资源，rollback 管事务，两者不能混为一件事

finally 在离开 try 时执行，适合关闭资源；rollback 撤销尚未提交的数据库变化。关闭连接不是业务撤销的同义词。课程最小事务实验的顺序是：

flowchart TD
    A[创建 Session] --> B[执行整笔订单]
    B -->|全部成功| C[commit]
    B -->|中途异常| D[rollback]
    C -->|提交失败| D
    C -->|提交成功| E[close]
    D --> E
    E --> F{本次是否发生异常}
    F -->|是| G[失败继续传给调用者]
    F -->|否| H[正常返回]

正常返回和异常传播分开核对。真实代码中 close 放在 finally，就算提交或回滚出错也尝试释放连接。更复杂的“清理失败掩盖原始异常”留作源码阅读，不要求今天设计完整资源框架。

用 contextmanager 写一次进入、一次退出

为什么学：商城一些资源边界只有“打开—使用—关闭”，为它们单独写类会重复。@contextmanager 将恰好 yield 一次的生成器变成上下文管理器；这次 yield 把资源交给 with，退出时再回到这里继续执行。

from contextlib import contextmanager
 
@contextmanager
def trace_operation():
    print("开始")
    try:
        yield "一次订单操作"
    finally:
        print("收尾")
 
with trace_operation() as label:
    print(label)

D01 的生成器可以产出多件商品，这里的生成器只交出一次资源。若 with 内抛错，异常会在 yield 处回到生成器；捕获后需要 raise 才能继续向外报告。

我补 commerce_ops/session_scope.py：先完成类协议，再用装饰器写等价边界。正常提交一次、失败回滚、始终关闭；失败不能被当作成功。标准库例捕获 BaseException 是为了在取消/中断时也清理并立即重新抛出，不是建议业务代码吞掉所有错误。

python check_d02_python.py

测试替身记录 commit/rollback/close，只证明调用顺序；真实 MySQL 仍要比较订单、明细和库存前后值。

与商城 MySqlUnitOfWork 对齐，但不混用提交策略

商城当前 UoW 在业务服务里显式 commit，退出时异常或未提交就 rollback，然后 close。小实验为了易观察，展示“正常退出自动 commit”。这两种都是可能的设计，但不能混着使用，否则会多提交或误提交。

我保留商城的显式提交策略，在现有 transaction_lab 中完成一笔订单一次提交；不拿本日最小上下文例覆盖整个 UoW。先找到真实 __enter__、__exit__ 和 commit，再运行第二项失败的回滚实验。

顺便解决昨天的提前停止：若只读取一条商品，用 with closing(iter_products(path)) as rows: 包住生成器；退出时调用生成器 close()，它会退出内部文件的 with。closing 只负责关闭，不会自动给文件导出提供数据库事务。

Service 与 Unit of Work：一次用例一次事务

我先回看刚才的半张订单：预留库存、订单和明细必须一起成功或一起回滚。如果每个数据访问函数自己提交，Service 就无法保证原子性。 Unit of Work（工作单元，简称 UoW）把这组操作放进同一个事务。当前实现把数据访问方法也放在 UoW 中，没有独立 crud.py 模块。

flowchart LR
    R[Router 解析 HTTP] --> S[CommerceService 业务用例]
    S --> P[UnitOfWork 协议]
    P -. MySQL 实现 .-> U[MySqlUnitOfWork]
    U --> DB[(MySQL)]
    P -. 测试实现 .-> F[Fake UoW]

文件与角色	负责什么	如何判断边界
web/router.py	解析输入、响应模型、调用 Service	不写 SQL、不管理事务
domain/services.py	库存、金额、状态与事务用例	抛业务异常，不返回 HTTPException
domain/ports.py	声明 UnitOfWork 需要的方法	只声明接口，不连接数据库
infrastructure/mysql/uow.py	Session、查询映射、提交回滚	一个用例一个工作单元
web/error_adapter.py	将业务异常翻译为 HTTP	稳定错误包络，不泄露 SQL

CRUD（Create/Read/Update/Delete，增删改查）描述操作类型；UoW 描述事务边界，二者不是互斥架构。原先的 Router → Service → 数据访问 → DB 职责仍成立，下面将它落实为当前文件中的接口。

先讲 Protocol，再看 with

Protocol 是 Python 的结构化类型协议：声明调用方需要哪些方法。对象具有这些方法就能被类型检查器识别为兼容，不要求继承这个协议；它不替代运行时校验。 下面是接口节选，省略其他方法，不能单独当作完整项目运行：

from typing import Protocol
 
class UnitOfWork(Protocol):
    def __enter__(self) -> "UnitOfWork": ...
    def __exit__(self, exc_type, exc, tb) -> None: ...
    def commit(self) -> None: ...

沿用 W01 文件读取时用过、今天已拆解的 with 机制，补充对象协议：进入 with 调 __enter__ 创建 Session，退出调 __exit__。MySQL 实现遇到异常或未 commit 会 rollback，随后 close。读操作同样关闭 Session。... 在协议中表示只声明方法，不能误以为业务已经实现。

入口先直接调用 Service

今天入口是 Python 脚本／已有 CLI，直接调用 CommerceService。上图 Router 和错误适配器是下次课 HTTP 接入位置，现在只识别边界，不实现 FastAPI 装饰器。先完成真实数据库服务和失败回滚，再接网络入口。

Service 与数据访问

下面是行为示意，名称与实际入口对应；完整边界由项目中的 ID 校验和业务异常负责：

CommerceService.get_order(order_id)
  校验 order_id
  with self.uow_factory() as uow:
    order = uow.get_order(order_id)
    不存在则抛 ResourceNotFound
    返回 order

MySqlUnitOfWork 的 get_order 使用 select(OrderModel)、joinedload(OrderModel.items) 和 unique().first()，将 ORM 记录映射成领域对象。joinedload 一次读取关联明细；unique 去除连接造成的重复订单行。HTTP 404 由错误适配器决定。

生命周期与代价

一次业务用例 → 创建 UoW/Session → 多次读取或修改
             → 写用例显式 commit
             → 异常或未提交则 rollback → close

收益是事务边界清楚、Fake 与 MySQL 可以使用相同业务逻辑；代价是多一份接口和装配代码。小脚本未必需要 UoW，本周已有跨表事务和独立测试，因此使用它。Fake 测试证明业务行为，真实锁行与数据库约束仍需 MySQL 集成测试。

---

先建立同步数据库服务，再讨论异步

我保留同步 SQLAlchemy Session 和 PyMySQL 调用：一次用例创建一个 Session，成功只提交一次，异常回滚，最后关闭。Session 不能由所有请求共享；连接池可由整个应用复用。commit 结束事务，flush 把变更发送给数据库但不等于最终提交。

今天验证两个商品的订单：第二项失败时，第一项库存、订单头和明细必须一起回滚。W03 再用两个真实连接验证竞争最后一件库存，不把内存 Fake 的通过当成 InnoDB 锁证据。

学生自己动手，AI 只能改你允许改的地方

我先保存当前可运行版本，再完成“第二条明细失败时验证所有表和库存恢复”。让 AI 一次只修改负责该行为的文件；修改前预测正常与失败结果，检查差异，再运行回归。每人需要解释一处接受或拒绝的建议。

错误与边界怎样验收

今天重点检查：循环内提交、全局 Session、Fake 冒充 MySQL。保留输入、预期、实际输出和修复后的结果，不能只贴全绿截图。未验证的外部依赖如实记为待验证。

留给下一阶段

我们用失败后的数据库结果证明了一次下单是完整动作。明天不重新写这套规则，只在服务外面增加 HTTP 入口，让其他程序能按同一份输入输出约定调用它。

官方资料

• Python 上下文管理器协议
• contextlib 与 contextmanager
• SQLAlchemy Session