Python SQLite批量导入：重复事件不重写，坏数据整批回滚

重复投递和批次中途失败，是小型数据导入脚本常见的两个问题。本文用 Python 标准库 sqlite3 做一个可运行的小实验：同一个事件 ID 重放时不增加记录；批次中的一条数据违反约束时，本批前面写入的记录也撤销。所有数据均为自建样例，仅在本机 Python 3.11.9 环境验证，没有连接线上数据库。

先定义导入语义

演示表包含 event_id 和 payload 两列。event_id 使用主键并显式声明 NOT NULL，payload 使用 NOT NULL 和非空长度约束。业务约定为“首次写入生效”：相同 ID 再次到达时保留旧内容，即使新内容不同也不覆盖。这只是幂等策略的一种选择；需要更新的业务不能直接照搬。

批次约定为全部成功或全部回滚。executemany 负责重复执行参数化语句，事务边界由 with conn 管理。这里明确使用 isolation_level=“DEFERRED”，让这个示例走 sqlite3 的传统隐式事务控制；不要在不了解影响时改成 isolation_level=None，否则自动提交会改变实验的前提。

完整示例

保存为 sqlite_batch_demo.py，执行 python sqlite_batch_demo.py。代码只使用标准库，数据库为内存数据库，连接关闭后演示数据消失。

"""Run with Python 3.11+, using only the standard library."""
import sqlite3


def import_batch(conn, rows):
    # Keep the exception outside this block so __exit__ can roll back.
    with conn:
        conn.executemany(
            """INSERT INTO events(event_id, payload) VALUES (?, ?)
            ON CONFLICT(event_id) DO NOTHING""",
            rows,
        )


def read_rows(conn):
    return conn.execute(
        "SELECT event_id, payload FROM events ORDER BY event_id"
    ).fetchall()


def main():
    conn = sqlite3.connect(":memory:", isolation_level="DEFERRED")
    try:
        conn.execute("""CREATE TABLE events (
            event_id TEXT PRIMARY KEY NOT NULL,
            payload TEXT NOT NULL CHECK(length(payload) > 0)
        )""")
        batch = [("evt-001", "first"), ("evt-002", "second")]
        import_batch(conn, batch)
        assert read_rows(conn) == batch
        print("PASS: normal batch committed")

        import_batch(conn, batch)
        assert read_rows(conn) == batch
        print("PASS: replay did not create duplicates")

        import_batch(conn, [("evt-001", "changed")])
        assert read_rows(conn) == batch
        print("PASS: existing payload kept (first write wins)")

        try:
            import_batch(conn, [("evt-003", "third"), ("evt-004", "")])
        except sqlite3.IntegrityError:
            assert read_rows(conn) == batch
            print("PASS: invalid row rolled back the whole batch")
        else:
            raise AssertionError("expected CHECK failure")
    finally:
        conn.close()


if __name__ == "__main__":
    main()

四个断言分别证明了什么

第一项验证正常批次提交，读回两条记录。第二项将同一批数据再导入一次，读回结果仍是两条。第三项给已有 ID 传入不同 payload，确认旧值保留，避免把“去重”误解成“更新”。

第四项先插入 evt-003，再插入 payload 为空的 evt-004。第二条触发 CHECK 约束，异常穿过 with conn 边界后，整批事务回滚，因此 evt-003 也不会留下。程序会输出四行 PASS；失败时由断言或数据库异常直接暴露问题。

异常应该在哪里捕获

示例把 IntegrityError 捕获放在 import_batch 调用外部。这样 with conn 能观察到异常并执行回滚。如果在事务内部捕获异常后直接吞掉，正常退出代码块可能提交此前成功的写入，破坏整批失败的预期。

ON CONFLICT(event_id) DO NOTHING 只针对指定主键冲突采取忽略策略。这里没有使用 INSERT OR IGNORE，因为示例希望 payload 的 NOT NULL、CHECK 等数据问题仍然报错。数据库约束负责兜底，业务层仍应提前校验字段和长度。

使用范围与后续改进

with conn 不负责关闭连接，所以代码在 finally 中显式 close。本函数还假定调用时连接上没有其他尚未提交的业务事务；否则它的提交和回滚会影响同一个连接上的事务内容。需要嵌套单元时，应另行设计 SAVEPOINT 边界。

主键去重只保证本表的重复写入语义，不能保证邮件发送、HTTP 调用等外部副作用只执行一次。真实系统可根据场景考虑事务外盒与消费端去重。大批量导入还需要明确内存占用、锁等待和批次大小：拆批后只能保证每一批原子性，不能继续声称整个文件全部成功或全部回滚。

这个实验最有价值的部分是可验证的失败路径：除了检查“成功写进去了”，还检查“失败时没有留下半批数据”。将这两类断言一起保留，后续调整去重策略或事务配置时才更容易发现行为变化。