Python sqlite3 批量写入：失败时回滚整批数据

一、批量接口不等于业务事务

导入物料清单时，一批记录可能包含重复编号。业务要求是“这一批全部写入，或者一条也不写”，但把列表交给 executemany()，并不能单凭这个调用形式得到保证。它会反复执行同一条参数化语句；执行到中间失败时，前面语句的结果是否提交，还取决于连接的事务配置和异常处理。

SQLite 默认的 ABORT 冲突处理会撤销发生约束错误的那条语句，但不会因此自动撤销同一事务里之前成功的语句。在没有显式事务、连接又处于 SQLite 自动提交状态时，先前完成的插入甚至已经提交。捕获异常后返回“导入失败”，也不会把这些记录自动还原。

因此，先确定业务提交边界，再选择批量接口。下面让一个函数完整拥有一次事务：开始、批量写入、提交；中途出现异常则回滚，并继续把异常交给调用方。示例只连接 :memory:，不读取或修改任何已有数据库文件。

二、明确连接配置，再写提交逻辑

Python 的事务控制接口存在版本差异。3.12 之前主要通过 isolation_level 控制模块何时隐式开始事务；设置为 None，模块不再隐式开启事务，可以自行发送 SQL 事务命令。3.12 新增 autocommit，只有它处于 LEGACY_TRANSACTION_CONTROL 模式时，isolation_level 才控制这套旧行为。

下面在提供新参数的版本中显式选择旧事务控制，再关闭隐式开启。这样不依赖新版本的默认值。这个选择服务于演示显式 SQL 边界，并不表示所有应用都应该使用旧模式；如果项目统一采用 autocommit=False，就应围绕该模式另行设计事务，不要直接混用本函数。

import sqlite3


def connect_demo():
    options = {"isolation_level": None}
    if hasattr(sqlite3, "LEGACY_TRANSACTION_CONTROL"):
        options["autocommit"] = sqlite3.LEGACY_TRANSACTION_CONTROL
    return sqlite3.connect(":memory:", **options)


def insert_batch(con, rows):
    if con.in_transaction:
        raise RuntimeError("caller already has an active transaction")
    con.execute("BEGIN")
    try:
        con.executemany(
            "INSERT INTO items(code, quantity) VALUES (?, ?)", rows
        )
        con.execute("COMMIT")
    except BaseException:
        if con.in_transaction:
            con.execute("ROLLBACK")
        raise

BEGIN 放在 try 前面：开始事务本身失败时，函数还没有取得一个应由它回滚的事务。提交则包含在 try 内；提交报错也不会被当成成功返回。异常分支先检查 in_transaction，因为某些 SQLite 错误可能已经使事务自动回滚，再无条件发送 ROLLBACK 反而会产生新的错误。

这里捕获 BaseException 只用于清理并原样继续抛出，也覆盖中断等情况；没有吞掉错误或擅自重试。回滚本身仍可能因底层故障失败，实际项目应同时保留原始异常和回滚异常，停止复用状态不明的连接。这个短示例不能代替完整的存储故障恢复策略。

SQL 中的 ? 用于绑定记录值，不通过字符串拼接生成 SQL。参数绑定处理的是值，不是动态表名或字段名；需要动态结构时，应单独使用可信白名单，而不是把任意标识符塞进占位符。

三、用中间一条失败验证整批回滚

将下面代码接在前一个代码块后运行。先提交一条基准记录，再让新批次的第二条与它冲突。检查的不只是捕获了异常，还包括第一条新记录没有留下、原记录没有改变，以及连接已经退出事务。

con = connect_demo()
try:
    con.execute("""
        CREATE TABLE items(
            code TEXT NOT NULL PRIMARY KEY,
            quantity INTEGER NOT NULL CHECK(quantity >= 0)
        )
    """)
    insert_batch(con, [("seed", 1)])
    try:
        insert_batch(con, [("new-a", 2), ("seed", 9), ("new-b", 3)])
    except sqlite3.IntegrityError:
        pass
    else:
        raise AssertionError("duplicate code should fail")

    assert con.execute("SELECT * FROM items").fetchall() == [("seed", 1)]
    assert not con.in_transaction
    insert_batch(con, [("new-a", 2), ("new-b", 3)])
    assert con.execute("SELECT COUNT(*) FROM items").fetchone()[0] == 3
    print("失败批次已回滚，正确批次已提交")
finally:
    con.close()

本文代码已在 macOS、Python 3.9.6、SQLite 3.51.0 的独立内存数据库中运行。完整验证还覆盖了数量 CHECK 约束失败、输入迭代器在产出一条记录后抛错，以及调用方已开启事务三种情况。前两项没有留下该批次的部分插入；最后一项在进入函数时被拒绝，调用方已有事务保持有效，未被本函数回滚。

3.12 及以上的连接分支依据官方参数语义编写，本次没有在这些版本实际运行，也没有进行断电或磁盘故障测试。约束示例不是完整业务校验：SQLite 普通表的类型行为、编号是否允许空文本、业务数量范围等仍需另行规定。

四、不要让事务边界悄悄扩大

本函数拒绝已经处于事务中的连接，是为了避免替上层提交或回滚不属于这批数据的改动。SQLite 的 BEGIN 事务不能直接嵌套；需要在外层事务中隔离一段操作时，应单独设计 SAVEPOINT 方案，明确谁拥有最终提交权，不能把入口检查简单删掉。

也不要在这段流程中混入 executescript()。在旧事务控制模式下，它可能先提交已有事务，然后执行脚本；脚本若需要原子性，必须按文档明确组织自己的事务。Python 从 3.6 起也不再因执行 DDL 而自动提交既有事务，不能沿用更早版本的印象推断边界。示例在批次开始前创建表，批次里只做参数化插入。

with con: 是另一个常见混淆点：连接上下文管理器用于处理事务的提交或回滚，并不负责关闭连接，也不会替所有配置自动开启事务。本文使用显式 SQL，所以也显式在 finally 关闭连接，不把两种控制方式叠在一起。

并发环境还需要处理锁竞争。SQLite 可以有多个读事务，但同一时刻只有一个写事务；BEGIN IMMEDIATE 可以更早尝试取得写事务，却不能消除锁冲突。本例使用普通 BEGIN，没有设计重试。调用方应先准备和校验输入，缩短持锁时间，再根据错误类型及业务幂等性决定是否重新提交完整批次，不要从失败行之后盲目续写。

对于大批量导入，拆批会改变业务承诺：每批单独提交时，后面失败不会撤销前面已提交的批次。是否允许部分成功，应由接口约定决定。事务能保护的是明确划定的数据库更新边界，外部文件、网络请求和其他系统的副作用不会随这次 SQLite 回滚一起撤销。

参考：Python 标准库 sqlite3 文档的事务控制、executemany()、参数绑定和连接上下文管理器章节；SQLite 文档的 Transaction、ON CONFLICT 与 SAVEPOINT 章节。