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

这个模式在出现多个可选步骤时，可以用 contextmanager 收拢 ROLLBACK TO 和 RELEASE，避免某个分支漏掉释放。下面仍只依赖 Python 3.11+ 标准库：

from contextlib import contextmanager
import re
import sqlite3


@contextmanager
def savepoint(connection: sqlite3.Connection, name: str):
    if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) is None:
        raise ValueError("invalid savepoint name")

    connection.execute(f"SAVEPOINT {name}")
    try:
        yield
    except BaseException:
        connection.execute(f"ROLLBACK TO {name}")
        connection.execute(f"RELEASE {name}")
        raise
    else:
        connection.execute(f"RELEASE {name}")

调用方只捕获预期的局部失败，外层事务仍可继续：

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

    try:
        with savepoint(connection, "optional_job"):
            connection.execute(
                "INSERT INTO jobs (name) VALUES (?)",
                ("prepare",),
            )
    except sqlite3.IntegrityError:
        print("optional job rolled back")

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

这里保留名称校验很重要：SQLite 的保存点名不能使用参数占位符，不能把外部输入直接插入 SQL。这个封装会重新抛出异常，因此是否忽略某类局部失败仍由调用方明确决定。

再补一个嵌套场景的边界：名称格式校验能避免把动态字符串直接拼进 SQL，但如果不同层复用同一个保存点名，SQLite 会按栈中最近的同名保存点处理 ROLLBACK TO name 和 RELEASE name，手工维护时很容易回滚或释放错层。

如果封装允许嵌套，建议由 helper 为每次调用生成唯一的内部名称，把业务标签与 SQL 标识符分开；同时加一个嵌套测试，确认内层失败只撤销内层写入，外层写入和后续 publish 仍然存在。例如在外层插入 prepare，内层插入 optional 后故意触发约束异常，捕获异常后再插入 publish，最后断言查询结果严格为 prepare、publish。这样能同时验证保存点栈的行为和异常后的可继续提交。

可以直接让 helper 生成只供 SQL 使用的内部名称，调用方不再传保存点名。下面示例仅依赖 Python 3.11+ 标准库，并用嵌套调用验证内层失败不会撤销外层写入：

from contextlib import contextmanager
import sqlite3
import uuid


@contextmanager
def savepoint(connection: sqlite3.Connection):
    name = f"sp_{uuid.uuid4().hex}"
    connection.execute(f"SAVEPOINT {name}")
    try:
        yield
    except BaseException:
        connection.execute(f"ROLLBACK TO {name}")
        connection.execute(f"RELEASE {name}")
        raise
    else:
        connection.execute(f"RELEASE {name}")


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

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

    with savepoint(connection):
        connection.execute("INSERT INTO jobs VALUES (?)", ("outer",))
        try:
            with savepoint(connection):
                connection.execute("INSERT INTO jobs VALUES (?)", ("inner",))
                connection.execute("INSERT INTO jobs VALUES (?)", ("prepare",))
        except sqlite3.IntegrityError:
            pass

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

rows = connection.execute(
    "SELECT name FROM jobs ORDER BY rowid"
).fetchall()
assert rows == [("prepare",), ("outer",), ("publish",)]
connection.close()

运行 python nested_savepoint.py，无输出即断言通过。uuid4().hex 只包含十六进制字符，再加固定字母前缀后可直接作为 SQLite 标识符；名称不接收外部输入，也不会在嵌套层之间复用。

再补一个会直接破坏保存点边界的 API：Python 3.11 默认事务模式下，Connection.executescript() 会在执行脚本前先提交当前待处理事务。因此不要把它放进这个 savepoint helper；保存点会随提交消失，脚本中前面已成功的语句也可能已经单独提交。

环境：Python 3.11+；Linux、macOS 或 Windows；仅使用标准库 sqlite3。下面是最小复现：

import sqlite3


connection = sqlite3.connect(":memory:")
connection.execute(
    "CREATE TABLE jobs (name TEXT NOT NULL UNIQUE)"
)
connection.execute(
    "INSERT INTO jobs VALUES (?)",
    ("prepare",),
)
connection.execute("SAVEPOINT optional_job")

try:
    connection.executescript(
        """
        INSERT INTO jobs VALUES ('optional');
        INSERT INTO jobs VALUES ('optional');
        """
    )
except sqlite3.IntegrityError:
    try:
        connection.execute("ROLLBACK TO optional_job")
    except sqlite3.OperationalError as error:
        print(error)

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

运行 python executescript_savepoint.py，预期输出：

no such savepoint: optional_job
[('prepare',), ('optional',)]

如果需要保留局部回滚语义，应在保存点内部使用 execute() 或 executemany()，让事务边界由外层代码统一管理。确实需要执行整段脚本时，应把脚本本身的事务设计作为独立流程，不要假设外层保存点仍然有效。