Python/SQLite：用 SAVEPOINT 让单批失败不拖垮整次导入

rookLv1楼主

107 次浏览2 条回复

批量导入通常希望保留已经成功的批次，同时让包含坏数据的一批整体回滚。SQLite 的 SAVEPOINT 可以在外层事务中建立局部回滚点；发生约束错误时先 ROLLBACK TO，再 RELEASE 结束该保存点，后续批次仍可继续。

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

创建 savepoint_demo.py：

import sqlite3

connection = sqlite3.connect(":memory:")
connection.execute(
    "CREATE TABLE jobs (name TEXT PRIMARY KEY, state TEXT NOT NULL)"
)

batches = [
    [("alpha", "ready"), ("beta", "ready")],
    [("gamma", "ready"), ("alpha", "duplicate")],
    [("delta", "ready")],
]

with connection:
    for number, rows in enumerate(batches, start=1):
        connection.execute("SAVEPOINT current_batch")
        try:
            connection.executemany(
                "INSERT INTO jobs(name, state) VALUES (?, ?)",
                rows,
            )
        except sqlite3.IntegrityError as error:
            connection.execute("ROLLBACK TO current_batch")
            connection.execute("RELEASE current_batch")
            print(f"batch {number}: skipped ({error})")
        else:
            connection.execute("RELEASE current_batch")
            print(f"batch {number}: imported")

for row in connection.execute("SELECT name, state FROM jobs ORDER BY name"):
    print(row)

connection.close()

运行：

python savepoint_demo.py

第二批中的 alpha 违反主键约束，因此同批先插入的 gamma 也会被撤销；第一批和第三批不受影响，最终只输出 alpha、beta、delta 三行记录。

这里 ROLLBACK TO 只回到保存点，并不会关闭保存点，所以异常分支仍要执行 RELEASE。外层 with connection: 负责最终提交；若循环之外再抛出未处理异常，外层事务仍会整体回滚。若业务要求任何一批失败都取消全部导入，则不应捕获后继续，而应让异常离开 with 块。

阿

阿线Lv1版主

52 天前

管理员核对：这里需要修正一处关键的事务边界。Python 3.11 的 with connection: 不会在进入代码块时主动执行 BEGIN。当前代码第一次执行的事务控制语句是 SAVEPOINT current_batch；当此时没有外层事务时，它会成为最外层保存点，而 RELEASE current_batch 会直接提交该事务。结果是每个成功批次实际分别提交，之后即使在 with 块内抛出未处理异常，也无法回滚已经释放的批次。

应在循环前显式开启外层事务，使各个保存点真正嵌套在其中：

with connection:
    connection.execute("BEGIN")
    for number, rows in enumerate(batches, start=1):
        connection.execute("SAVEPOINT current_batch")
        try:
            connection.executemany(
                "INSERT INTO jobs(name, state) VALUES (?, ?)",
                rows,
            )
        except sqlite3.IntegrityError as error:
            connection.execute("ROLLBACK TO current_batch")
            connection.execute("RELEASE current_batch")
            print(f"batch {number}: skipped ({error})")
        else:
            connection.execute("RELEASE current_batch")
            print(f"batch {number}: imported")

这样 RELEASE 只结束当前批次的保存点，退出 with 时才提交外层事务；若未处理异常离开该代码块，外层事务才会整体回滚。建议作者修正文中“外层事务负责最终提交”的说明和示例。

hiLv1

52 天前

阿线Lv152 天前#1
管理员核对：这里需要修正一处关键的事务边界。Python 3.11 的 with connection: 不会在进入代码块时主动执行 BEGIN。当前代码第一次执行的事务控制语句是 SAVEPOINT current_batch；当此时没有外层事务时，它会成为最外层保存点，而 RELEASE current_batch 会直接提交该事务。结果是每个成功批次实际分别提交，之后即使在 with 块内抛出未处理异常，也无法回滚已经释放的批次。

应在循环前显式开启外层事务，使各个保存点真正嵌套在其中：
with connection:
    connection.execute("BEGIN")
    for number, rows in enumerate(batches, start=1):
        connection.execute("SAVEPOINT current_batch")
        try:
            connection.executemany(
                "INSERT INTO jobs(name, state) VALUES (?, ?)",
                rows,
            )
        except sqlite3.IntegrityError as error:
            connection.execute("ROLLBACK TO current_batch")
            connection.execute("RELEASE current_batch")
            print(f"batch {number}: skipped ({error})")
        else:
            connection.execute("RELEASE current_batch")
            print(f"batch {number}: imported")
这样 RELEASE 只结束当前批次的保存点，退出 with 时才提交外层事务；若未处理异常离开该代码块，外层事务才会整体回滚。建议作者修正文中“外层事务负责最终提交”的说明和示例。

可以再用 connection.in_transaction 做一个最小回归检查，直接验证 RELEASE 后外层事务仍然存在。

环境：Python 3.11+；仅使用标准库 sqlite3。

import sqlite3

connection = sqlite3.connect(":memory:")
connection.execute("CREATE TABLE jobs (name TEXT PRIMARY KEY)")

try:
    with connection:
        connection.execute("BEGIN")
        connection.execute("SAVEPOINT current_batch")
        connection.execute("INSERT INTO jobs VALUES ('alpha')")
        connection.execute("RELEASE current_batch")

        assert connection.in_transaction
        raise RuntimeError("abort outer transaction")
except RuntimeError:
    pass

count = connection.execute("SELECT COUNT(*) FROM jobs").fetchone()[0]
assert count == 0
print("outer rollback verified")

若删掉显式的 BEGIN，最外层 SAVEPOINT 被 RELEASE 后，connection.in_transaction 会变为 False，已插入的行也不会被随后离开 with 的异常撤销。把这两个断言留在示例测试里，可以防止以后调整事务代码时重新引入边界问题。

返回主题列表