18 数据操作之更新 —— 乐观锁：让数据库帮你检测并发冲突

前置阅读: 16-数据操作之查询、17-数据操作之新增
关键词: update, where, version, 乐观锁, rowcount
难度: ★★★★☆

场景导入：两个人同时编辑同一本书，谁来"说了算"？

新增是把新对象写入数据库，冲突主要是唯一键重复——第 17 章的 IntegrityError 处理已经覆盖。但更新的并发冲突更微妙：用户 A 和用户 B 同时打开同一本书的编辑页面，A 改了标题提交，B 随后也改了状态提交——B 的提交是基于过时的数据，可能把 A 的修改悄然覆盖。

这就是经典的"丢失更新"（lost update）问题。悲观锁的解法是 SELECT ... FOR UPDATE——读的时候就锁住行，别人等。乐观锁的解法是给每行加一个 version 列——更新时带上版本号条件，数据库帮你检测冲突。本期把第 06 章占位的 PUT /books/{book_id} 改造为带乐观锁的真实端点，使用 SQLAlchemy Core 层的 update(Book).where(...).values(...) 写法。

原理解析：Core 层 update + rowcount = 乐观锁

更新在 SQLAlchemy 中有两条路径。ORM 路径是"读取对象 → 修改属性 → commit"——适合需要保留对象身份、触发 ORM 钩子的场景。Core 层路径是 update(Book).where(...).values(...)——直接发 UPDATE 语句，绕过身份映射和脏检测，更适合按条件批量更新和乐观锁场景。

本期选择 Core 层路径，因为乐观锁天然需要在 UPDATE 时附加 WHERE version = ? 条件——Core 层可以直接在 where() 里写 Book.version == version，语义精确：

（图注：update(...).where(Book.id == ..., Book.version == ...) 构造的语句在 await db.execute 时一次性发送，rowcount 决定走 commit 流程还是抛锁定冲突。）

乐观锁的核心机制可以浓缩为一条 SQL：

UPDATE books
   SET title = '新标题', status = 'BORROWED', version = version + 1
 WHERE id = 3
   AND version = 2

关键在于 SET version = version + 1——这是在数据库端基于当前值自增。绝对不能在 Python 侧预先计算 version + 1，否则并发场景下两个请求可能都算出 version=3 然后写入，乐观锁退化为覆盖更新。在 SQLAlchemy 中，values(version=Book.version + 1) 中的 Book.version + 1 不是整数，而是 SQLAlchemy 的 BinaryExpression——它被翻译为 SQL 的 version = version + 1，由 MySQL 在执行 UPDATE 时原子计算。

result.rowcount 是 MySQL 返回的受影响行数。在乐观锁场景下，rowcount == 0 有两种可能：id 不存在（该行从未存在），或 version 不匹配（被并发更新改过）。本系列不区分两者，统一抛 OptimisticLockError(book_id)，由第 08 章的异常处理器映射为 409 Conflict。如果需要区分 404 vs 409，可以额外 SELECT 一次，但通常得不偿失。

await db.commit() 在 rowcount == 1 后提交事务。Core 层 update() 不经过 Session.dirty，但 SQLAlchemy 在 commit 阶段仍会走 ORM 钩子（本系列未启用 version_id_col）。异常路径上，OptimisticLockError 抛出后由 get_async_db 的 async with 退出阶段自动 rollback。

（图注：Core 层 update 的执行流程比 ORM 路径短，但每一步都是显式 await，事务边界与返回值边界清晰。）

代码实现：update_book + OptimisticLockError + BookUpdate

DAO 层追加 update_book。values 字典里混合了字面量（title、status）和 SQL 表达式（Book.version + 1）——两者可以共存：

# app/dao/book_dao.py（增量追加）
from sqlalchemy import update
from sqlalchemy.ext.asyncio import AsyncSession

from app.errors import OptimisticLockError
from app.models.book import Book, BookStatus


async def update_book(
    db: AsyncSession,
    book_id: int,
    *,
    version: int,
    title: str | None = None,
    status: BookStatus | None = None,
) -> Book:
    """按 version 乐观锁更新图书；rowcount == 0 表示版本不匹配或已删除。"""
    values: dict = {"version": Book.version + 1}
    if title is not None:
        values["title"] = title
    if status is not None:
        values["status"] = status
    stmt = (
        update(Book)
        .where(Book.id == book_id, Book.version == version)
        .values(**values)
    )
    result = await db.execute(stmt)
    if result.rowcount == 0:
        raise OptimisticLockError(book_id)
    await db.commit()
    refreshed = await db.get(Book, book_id)
    if refreshed is None:
        raise OptimisticLockError(book_id)
    return refreshed

None 哨兵区分"未提供"与"显式置空"——调用方不传 title 时，该字段不进 SET 子句，行内原值保留。await db.get(Book, book_id) 拉回最新行，响应里 version 是更新后的值，客户端可以基于此发起下一次更新。

领域异常类：

# app/errors.py（新增）
class OptimisticLockError(Exception):
    """乐观锁冲突：id 对应的记录版本不匹配或不存在。"""

    def __init__(self, book_id: int) -> None:
        super().__init__(f"book {book_id} version mismatch")
        self.book_id = book_id

Schema 层定义 BookUpdate——version 是必填字段（乐观锁的核心），title 和 status 是可选项：

# app/schemas/book.py（增量）
from pydantic import BaseModel, Field

from app.models.book import BookStatus


class BookUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=128)
    status: BookStatus | None = None
    version: int = Field(gt=0)

Pydantic v2 下 Field(default=None, min_length=...) 对 None 不做长度校验，只在实际提供字符串时校验——这是正确的行为。路由层追加 PUT /books/{book_id} handler：

# app/api/books.py（追加路由）
from app.dao import book_dao
from app.schemas.book import BookUpdate


@router.put("/{book_id}")
async def update_book_route(book_id: int, payload: BookUpdate, db: DBDep) -> Book:
    """按 version 乐观锁更新图书；冲突由 OptimisticLockError 处理器映射为 409。"""
    return await book_dao.update_book(
        db,
        book_id,
        version=payload.version,
        title=payload.title,
        status=payload.status,
    )

避坑指南

• Book.version + 1 必须是 SQL 表达式，不是 Python 整数。 写成 values["version"] = version + 1 会把 Python 侧的 version + 1 预设值绑定为常量参数——翻译为 SET version = 3，乐观锁自增语义丢失。

• pool_pre_ping 防长连接关闭。MySQL 默认 wait_timeout 后关闭空闲连接，低流量时段取出的连接可能已失效。第 13 章设置的 pool_pre_ping=True 让每次借连接前发 SELECT 1 探测，失效连接自动重建。

• InnoDB 行锁行为。UPDATE ... WHERE id = ? AND version = ? 在执行阶段按 id 走聚簇索引定位行并加排他锁。两个并发事务中，先拿到锁的正常提交（rowcount=1），后者阻塞到锁释放后被唤醒，但此时 version 已被前者自增，WHERE 不再命中——rowcount=0 立即返回。这是"锁 + 校验"的组合：锁保证同时只有一个事务写，version 校验保证不会基于过时数据更新。

• rowcount == 0 不区分"行不存在"和"版本不匹配"。如果需要精细化错误码（404 vs 409），可在 DAO 内追加一次 SELECT COUNT(*) WHERE id = ?。大多数场景不需要——客户端先 GET 拿到 version，PUT 失败大概率是版本冲突。

面试 QA

Q1 [原理]: 乐观锁的实现细节是什么？rowcount == 0 一定意味着冲突吗？

乐观锁的核心是 WHERE version = ? AND SET version = version + 1 这对组合。事务 A 和 B 同时读到 version=1，A 先提交后 version 变为 2，B 再提交时 WHERE version=1 不命中，rowcount=0，DAO 抛 OptimisticLockError。

rowcount == 0 不一定是版本冲突——id 不存在也会返回 0。在生产实践中，客户端此前已通过 GET 验证 id 存在，rowcount==0 大概率是版本冲突。不区分两者可以简化异常处理，统一返回 409。

异步栈下需要注意：update(Book).where(...).values(...) 是同步构造，await db.execute(stmt) 才是 I/O 边界，await db.commit() 和 await db.get() 也必须加 await。缺失任何一处都会在运行时抛 TypeError。

Q2 [项目]: Alembic 异步迁移如何编写？本章为什么不展开？

Alembic 异步迁移由第 14 章的 env.py 改造覆盖——核心是用 create_async_engine + run_sync 把同步迁移入口适配到异步连接。本章不展开是因为聚焦于"运行时更新路径"，与迁移脚本是两条独立轨道。version 列的迁移随 Book 模型首次建表时一并定义，后续新增列（如 updated_at）可在新的 migration 文件中追加。

小结

本期把 PUT /books/{book_id} 改造为带乐观锁的 Core 层更新：update(Book).where(Book.id == book_id, Book.version == version).values(version=Book.version + 1, ...) 构造语句 → await db.execute(stmt) 执行 → result.rowcount 校验 → await db.commit() 提交 → await db.get(Book, book_id) 回填最新状态。Book.version + 1 作为 SQL 表达式传入——不是 Python 预计算——是乐观锁自增的关键。

下一篇《19 数据操作之删除》将进入删除场景，讨论软删除（UPDATE ... SET deleted_at = :now）与查询侧过滤的协同——行不消失，只是被标记。