用 Python 标准库演示 SQLite 事务：一次失败，整批回滚

把一个任务拆成多条数据库写入时，最怕前几条成功、最后一条失败，留下半成品。Python 自带的 sqlite3 可以在本地演示事务的原子性，不需要安装数据库服务。本文用一个小型任务表验证两件事：唯一约束怎样阻止重复标题，以及 with conn: 怎样在异常时回滚整批写入。

场景与表结构

我们创建 tasks 表，包含任务 ID、标题和优先级。标题设为 UNIQUE，优先级只允许 1 到 3。示例使用内存数据库 :memory:，程序退出后数据自动消失，适合复现实验；如果改为文件路径，数据就会写到本地 SQLite 文件。

下面是完整可运行的 Python 3 脚本，保存为 sqlite_transaction_demo.py：

#!/usr/bin/env python3
"""Demonstrate atomic writes and unique constraints with sqlite3."""

import sqlite3


def main() -> None:
    conn = sqlite3.connect(":memory:")
    try:
        conn.execute(
            "CREATE TABLE tasks ("
            "id INTEGER PRIMARY KEY, "
            "title TEXT NOT NULL UNIQUE, "
            "priority INTEGER NOT NULL CHECK(priority BETWEEN 1 AND 3))"
        )

        try:
            with conn:
                conn.execute(
                    "INSERT INTO tasks (title, priority) VALUES (?, ?)",
                    ("备份", 2),
                )
                conn.execute(
                    "INSERT INTO tasks (title, priority) VALUES (?, ?)",
                    ("备份", 3),
                )
        except sqlite3.IntegrityError as exc:
            print(f"重复任务导致整批回滚：{exc}")

        count = conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0]
        print(f"失败后条数：{count}")

        with conn:
            conn.execute(
                "INSERT INTO tasks (title, priority) VALUES (?, ?)",
                ("检查", 1),
            )
            conn.execute(
                "INSERT INTO tasks (title, priority) VALUES (?, ?)",
                ("发布", 2),
            )

        rows = conn.execute(
            "SELECT id, title, priority FROM tasks ORDER BY id"
        ).fetchall()
        print(f"成功后的任务：{rows}")
    finally:
        conn.close()


if __name__ == "__main__":
    main()

运行 python3 sqlite_transaction_demo.py。我在本地运行得到：

重复任务导致整批回滚：UNIQUE constraint failed: tasks.title
失败后条数：0
成功后的任务：[(1, '检查', 1), (2, '发布', 2)]

为什么第一批一条也没留下？

第一条“备份”插入本来可以成功；第二条标题相同，触发 UNIQUE 约束并抛出 sqlite3.IntegrityError。异常离开 with conn: 代码块时，连接会回滚当前事务，因此查询得到的条数是 0。第二批没有异常，离开代码块时提交，最终留下“检查”和“发布”两条记录。

with conn: 管理的是事务提交或回滚，不会自动关闭连接，所以脚本仍在 finally 中调用 conn.close()。捕获异常应放在 with conn: 的外侧；若在代码块内部吞掉异常，块正常结束时仍可能提交已执行的语句。需要更复杂的错误恢复时，应明确设计事务或保存点边界。

两个容易忽略的边界

第一，SQL 值使用 ? 占位符，实际内容作为参数单独传入。这样可以正确处理引号等字符，也避免把用户输入直接拼成 SQL 语句。占位符用于值，不能拿来替代任意表名或列名。

第二，约束由数据库执行，不依赖“先查再插”。先查询是否有同名任务，再插入，看似可以避免重复，但并发写入时两次查询都可能看到“尚不存在”。UNIQUE 约束才是最终防线。业务若需要“重复时跳过”，可以有意识地采用 INSERT ... ON CONFLICT(title) DO NOTHING；本例故意保留异常，以展示整批回滚。

事务适合把一组必须同时成功的写入放在一起。实际项目还要根据持久化要求选择数据库文件、备份方案，并在测试中验证失败路径是否真的保持数据一致。