跳到主要内容

5.2 事务处理

转账转到一半程序崩了，钱扣了但没到账——事务就是防止这种惨剧的机制。本节讲透 SQLAlchemy 中事务的正确姿势。

一、什么是事务？​

事务（Transaction）= 一组必须"同生共死"的数据库操作。

经典例子——转账：

① 从张三账户扣 100 元
② 给李四账户加 100 元

如果 ① 成功后程序崩溃、② 没执行，100 元就凭空消失了。事务保证：要么两步都成功（commit），要么都不生效（rollback），不存在中间状态。

事务的四大特性（ACID）了解即可：原子性（全成或全败）、一致性（数据始终合法）、隔离性（并发事务互不干扰）、持久性（提交后断电也不丢）。

二、Session 的事务模型：你一直在用事务​

关键认知：Session 的所有操作天然就在事务里。

with SessionLocal() as session:
    session.add(user1)        # ┐
    user2.age = 30            # ├─ 同一个事务
    session.delete(user3)     # ┘
    session.commit()          # 三个操作一起生效
  • 从第一个数据库操作开始，Session 自动开启事务
  • commit() 提交事务；rollback() 撤销事务内所有未提交操作
  • commit 后再操作，会自动开启新的事务

所以"把相关操作放进同一次 commit"不只是性能优化，更是正确性保证。

三、转账示例：事务的标准写法​

from decimal import Decimal
from sqlalchemy import String, Numeric
from sqlalchemy.orm import Mapped, mapped_column

class Account(Base):
    __tablename__ = "accounts"

    id: Mapped[int] = mapped_column(primary_key=True)
    owner: Mapped[str] = mapped_column(String(50))
    balance: Mapped[Decimal] = mapped_column(Numeric(12, 2), default=0)

写法1：try / except / rollback（基础版）​

def transfer(session, from_id: int, to_id: int, amount: Decimal):
    try:
        from_acc = session.get(Account, from_id)
        to_acc = session.get(Account, to_id)

        if from_acc.balance < amount:
            raise ValueError("余额不足")

        from_acc.balance -= amount
        to_acc.balance += amount
        session.commit()               # 两笔变动一起生效
    except Exception:
        session.rollback()             # 任何一步出错，全部撤销
        raise                          # 继续抛出，让调用方知道失败了

写法2：session.begin() 上下文（推荐）​

with session.begin(): 块结束自动 commit、异常自动 rollback，不用手写 try/except：

def transfer(session, from_id: int, to_id: int, amount: Decimal):
    with session.begin():
        from_acc = session.get(Account, from_id)
        to_acc = session.get(Account, to_id)

        if from_acc.balance < amount:
            raise ValueError("余额不足")     # 抛异常 → 自动 rollback

        from_acc.balance -= amount
        to_acc.balance += amount
    # 走到这里 = 已自动 commit

也可以创建 session 时一步到位：

with SessionLocal() as session, session.begin():
    ...  # 增删改，结束自动提交

⚠️ 用了 session.begin() 就不要在块内再手动 session.commit()，会报错"事务已由上下文管理"。

四、rollback 之后 Session 还能用吗？​

能。rollback 会把 Session 恢复到干净状态，可以开始新的事务：

with SessionLocal() as session:
    try:
        session.add(User(email="duplicate@example.com"))   # 假设撞了唯一约束
        session.commit()
    except IntegrityError:
        session.rollback()

    # session 依然可用，继续干别的
    users = session.scalars(select(User)).all()

但注意：rollback 后，之前 add 的未提交对象会回到游离态、已加载对象的未提交修改会丢失。需要重试的话，重新构造对象。

📌 复习一个 3.1 节的要点：commit 失败后必须 rollback，否则 Session 卡在"故障"状态，后续操作全部抛 PendingRollbackError。

五、嵌套场景：函数里该不该 commit？​

新手常见困惑：写了个 create_user() 函数，里面要不要 commit？

最佳实践：底层函数只操作、不提交；由最外层的调用方统一提交。

# ✅ 推荐：CRUD 函数不 commit（最多 flush），提交权交给调用方
def create_user(session, name: str, email: str) -> User:
    user = User(name=name, email=email)
    session.add(user)
    session.flush()          # 需要立刻拿 id 的话 flush 一下
    return user

def create_post(session, user: User, title: str) -> Post:
    post = Post(title=title, author=user)
    session.add(post)
    return post

# 调用方掌控事务边界：用户和文章要么都建好，要么都不建
with SessionLocal() as session, session.begin():
    user = create_user(session, "张三", "zs@example.com")
    create_post(session, user, "第一篇")

如果每个小函数内部都 commit，就把一个完整业务拆成了多个独立事务，中途失败会留下"半成品数据"。

一句话原则：谁代表完整业务，谁 commit。（在 FastAPI 里，"一个请求 = 一个事务"，第 6 章的依赖注入正是这么设计的。）

六、并发问题初探：两个请求同时扣库存​

# 库存只剩 1 件，两个请求同时执行：
stock = session.get(Product, 1).stock    # 两边都读到 1
if stock > 0:
    product.stock -= 1                   # 两边都扣成功 → 超卖！
    session.commit()

两种经典解法，先混个脸熟：

悲观锁：SELECT ... FOR UPDATE​

with session.begin():
    product = session.scalar(
        select(Product).where(Product.id == 1).with_for_update()
    )        # 锁住这一行，其他事务读它会阻塞等待
    if product.stock > 0:
        product.stock -= 1

乐观做法：原子 UPDATE​

result = session.execute(
    update(Product)
    .where(Product.id == 1, Product.stock > 0)     # 条件写进 SQL，由数据库保证原子性
    .values(stock=Product.stock - 1)
)
session.commit()
if result.rowcount == 0:
    print("库存不足，扣减失败")

💡 SQLite 对并发支持较弱（with_for_update 无效），这些技巧主要用于 MySQL/PostgreSQL。新手阶段知道"并发扣减要用锁或原子 UPDATE"即可，遇到再深究。

📝 本节小结​

  • 事务 = 一组同生共死的操作；Session 的操作天然在事务中，commit 定边界
  • 推荐 with session.begin():——自动 commit、异常自动 rollback
  • commit 失败必须 rollback，否则 Session 罢工
  • 底层函数不 commit（最多 flush），谁代表完整业务谁 commit
  • 并发扣减用 with_for_update()（悲观锁）或条件原子 UPDATE

下一节解决"改表结构"的难题 → 5.3 Alembic 数据库迁移