Python sqlite3 事务为什么回滚不了：commit、异常处理与连接边界

订单导入脚本明明捕获了异常，数据库里却留下了半条订单：主表已经写入，明细表没有写完，重跑后又遇到唯一键冲突。Python 的 sqlite3 里，回滚失效通常不是 rollback() 这个方法不存在，而是提交时机、异常边界和连接对象没有放在同一个事务里。

不少Python开发者在使用标准库自带的sqlite3模块时，都遇到过明明写了rollback操作，异常触发后之前写入的数据却没被撤销的情况，这类问题几乎都和连接边界处理不当、异常捕获逻辑错误、提交权限分散有关。

先确认：到底是哪一部分没有回滚

sqlite3的回滚操作，只会作用在当前连接里还没提交的未完结事务上。它没法撤销已经执行完提交的写入操作，更没法把其他连接已经成功落盘的数据删掉。排查这类问题的第一步别上来就改代码改成插一行就提交，这么做反而会把原本的事务一致性问题彻底掩盖，后续更难定位根因。

你完全可以先搭个最小可复现的、用临时文件存储的数据库做测试，把建表逻辑、两次写入操作和主动抛出的异常都放到同一个函数里，直接对比异常触发前后表里的总数据行数，很快就能定位是不是回滚没生效。

import sqlite3

db = sqlite3.connect(":memory:")
db.execute("create table order_item (id integer primary key, name text not null)")

try:
    db.execute("insert into order_item(id, name) values (?, ?)", (1, "keyboard"))
    db.execute("insert into order_item(id, name) values (?, ?)", (1, "mouse"))
    db.commit()
except sqlite3.IntegrityError as exc:
    db.rollback()
    print(type(exc).__name__, db.execute("select count(*) from order_item").fetchone()[0])
finally:
    db.close()

第二次写入触发主键冲突后，查询结果应为 0。如果结果是 1，重点检查之前是否有单独的 commit()，或者两次写入是否实际使用了不同的连接。

把事务边界和业务动作绑在一起

后续写生产代码的时候，更稳妥也更容易维护的方案是让业务入口的调用方统一持有数据库连接，底层各个业务写入的工具函数只负责执行SQL，完全不做擅自提交的操作。这样订单头、关联订单行、库存扣减记录这整套逻辑要么全部写入成功，要么出问题时在入口同一处统一回滚，不会出现部分写入部分失败的脏数据。

def add_order(db, order_id, customer):
    db.execute(
        "insert into orders(id, customer) values (?, ?)",
        (order_id, customer),
    )


def add_item(db, order_id, sku, quantity):
    db.execute(
        "insert into order_item(order_id, sku, quantity) values (?, ?, ?)",
        (order_id, sku, quantity),
    )


def import_one(db, order_id, customer, sku, quantity):
    add_order(db, order_id, customer)
    add_item(db, order_id, sku, quantity)


with sqlite3.connect("orders.db") as db:
    import_one(db, 1001, "Lin", "KB-01", 2)

这里的关键不是把所有函数都写成方法，而是明确谁拥有提交权。底层函数如果在 add_order() 结束时提交，后面的明细写入失败就无法撤回订单头。

异常捕获不能把失败伪装成成功

很多开发者最容易踩的坑，就是在事务的执行逻辑内部捕获到异常之后，只打一行日志就直接return返回上层，完全没触发回滚操作：

def import_bad(db, row):
    try:
        db.execute("insert into orders(id, customer) values (?, ?)", row)
    except sqlite3.IntegrityError as exc:
        print("skip:", exc)
    return True

调用方看到 True，可能继续处理下一条并提交当前连接。更稳妥的是让异常继续向事务拥有者传播，或者返回明确的失败结果，同时由拥有者决定是否回滚：

def import_one_checked(db, row):
    try:
        db.execute("insert into orders(id, customer) values (?, ?)", row)
    except sqlite3.IntegrityError:
        raise


try:
    with sqlite3.connect("orders.db") as db:
        import_one_checked(db, (1001, "Lin"))
        db.execute("insert into order_item(order_id, sku, quantity) values (?, ?, ?)",
                   (1001, "KB-01", 2))
except sqlite3.IntegrityError as exc:
    print("batch failed:", exc)

异常类型也要分开处理，不能一概而论：碰到唯一键冲突、非空字段为空、外键约束校验失败这类情况，大多属于输入参数或者存量数据本身有问题；碰到磁盘无写入权限、数据库文件损坏这类情况，则属于运行环境层面的故障。两类场景触发后都要执行回滚，但日志记录方式、告警级别、后续的重试策略不能全部混成一个笼统的“操作失败”返回。

with 连接的提交和回退规则

with sqlite3.connect(...) 退出时，会根据代码块是否带异常来提交或回滚事务；它不会替你关闭连接。短批处理可以使用这种边界，但不要在块内再开启一套相互矛盾的提交策略。

with sqlite3.connect("orders.db") as db:
    db.execute("insert into audit(event) values (?)", ("order-import",))
    # 这里发生异常，离开 with 时当前事务回退

# db 仍然是一个连接对象；长生命周期程序应在合适的生命周期结束处 close()

如果你的应用把sqlite3连接存在线程局部变量里复用，还要额外确认这个连接没有被其他线程跨线程误用。连接的生命周期越长，越要跟着记录对应的业务批次号、事务开始时间、本次写入成功的行数、最终是提交还是回滚的状态，只靠函数返回了一个“执行成功”的标识，根本没法判断事务是不是真的已经安全落盘。

嵌套调用时，谁负责提交必须先说清

当你一个外层事务要调用好几个内层的业务函数时，内层函数绝对不能随便提交或者回滚整个连接的全局状态。如果业务确实需要支持嵌套的局部回退边界，可以用sqlite3的保存点能力，但要清楚保存点只是当前大事务里的一个局部回退位置，没法代替最外层的最终提交操作。

with sqlite3.connect("orders.db") as db:
    db.execute("savepoint item_check")
    try:
        db.execute("insert into order_item(order_id, sku, quantity) values (?, ?, ?)",
                   (1001, "KB-01", 2))
        db.execute("release item_check")
    except sqlite3.IntegrityError:
        db.execute("rollback to item_check")
        db.execute("release item_check")
        raise

保存点的命名规则、释放顺序、异常向上传播的逻辑，全部要写进单元测试覆盖到。别把保存点当成随便吞掉错误的开关，就算局部写入失败用保存点回退了，只要外层最终执行了提交，此时整个数据库的状态也必须完全符合对应的业务约束要求。

用三组证据验收回滚是否真的生效

写测试用例的时候别只断言代码运行完没抛异常，至少要覆盖三类核心场景：

• 成功批次：订单头、订单行和审计记录的数量一起增加。
• 约束失败批次：事务前的行数与事务后的行数一致，失败批次号被记录。
• 连接复用批次：同一连接连续处理两批，第一批提交后第二批失败，不应影响第一批。

如果使用文件数据库，再把数据库文件复制到临时目录测试，避免测试数据与开发环境共享。检查查询结果时使用明确的 select count(*) 和业务主键，不要拿日志中的“开始处理”当作成功证据。

def count_items(db, order_id):
    return db.execute(
        "select count(*) from order_item where order_id = ?",
        (order_id,),
    ).fetchone()[0]


assert count_items(db, 1001) == 2
# 失败批次应保持失败前的数量，而不是留下半条记录

常见问题与边界

调用 rollback() 后为什么仍然能查到一行？

碰到明明写了rollback但是数据还在的情况，先排查这条数据是不是在本次事务启动之前就已经被提交过了，再确认你事后查询数据用的连接，和你执行写入回滚的是不是同一个连接。还有一类非常高发的测试坑，是测试代码在执行回滚之前就把数据库里的结果读到了本地缓存变量里，之后拿这个缓存的旧结果当成数据库的真实状态，误判回滚没生效。

每条 SQL 都 commit() 会更安全吗？

这种每插一行就立刻提交的写法，确实能把单条写入失败的影响范围缩到最小，但直接破坏了多步关联业务的一致性，只有当每一条写入本身就是完全独立的业务动作时，才适合这么写。

异常被捕获后一定要手动 rollback() 吗？

如果异常从 with sqlite3.connect() 代码块中离开，上下文管理器会处理当前事务；如果你在块内捕获并继续执行，就必须明确决定回滚、重试还是把失败交给外层。

总结：先收紧连接边界，再谈回滚

Python sqlite3的回滚有效性校验，核心其实就三件事：同一批关联业务的全部写入操作复用同一个数据库连接，提交权限集中在事务的拥有者手里统一调度，操作失败时直接查数据库的实际行数和约束校验结果做反向核对。把这三点落到自动化测试用例里之后，回滚是否生效就再也不用靠日志里一句模糊的“已处理”来判断，每一次执行都有可复现的明确证据。