Python sqlite3：用 SAVEPOINT 回滚事务中的局部失败

9 次浏览0 条回复

一个事务里有必做步骤和可失败的可选步骤时，直接回滚整个事务往往过重。SQLite 的 SAVEPOINT 可以建立局部回滚点：可选步骤失败后只撤销该段操作，外层事务仍可继续。

环境：Python 3.11+；Linux、macOS 或 Windows；仅使用标准库 sqlite3。

创建 savepoint_demo.py：

import sqlite3


connection = sqlite3.connect(":memory:")
connection.execute(
    "CREATE TABLE jobs (name TEXT NOT NULL UNIQUE)"
)

with connection:
    connection.execute(
        "INSERT INTO jobs (name) VALUES (?)",
        ("prepare",),
    )

    connection.execute("SAVEPOINT optional_job")
    try:
        # 重复名称触发 UNIQUE 约束，用来模拟可选步骤失败。
        connection.execute(
            "INSERT INTO jobs (name) VALUES (?)",
            ("prepare",),
        )
    except sqlite3.IntegrityError:
        connection.execute("ROLLBACK TO optional_job")
        connection.execute("RELEASE optional_job")
        print("optional job rolled back")
    else:
        connection.execute("RELEASE optional_job")

    connection.execute(
        "INSERT INTO jobs (name) VALUES (?)",
        ("publish",),
    )

rows = connection.execute(
    "SELECT name FROM jobs ORDER BY rowid"
).fetchall()
print(rows)
connection.close()

运行：

python savepoint_demo.py

预期输出：

optional job rolled back
[('prepare',), ('publish',)]

这里有两个关键点。第一，ROLLBACK TO optional_job 只把事务恢复到保存点创建时，并不会结束外层事务，所以后面的 publish 仍能写入。第二，回滚到保存点后还要执行 RELEASE optional_job 来移除保存点；如果可选步骤成功，else 分支也同样需要释放它。

示例只捕获预期的 sqlite3.IntegrityError。其他异常会继续向外传播，由 with connection: 回滚整个外层事务，避免把程序错误误当成可以忽略的局部失败。保存点名称不能通过 SQL 参数占位符传入；名称来自动态输入时应先做白名单映射，而不是直接拼接。