导入程序先写入批次记录，再调用公共函数写明细，最后校验失败抛出异常。代码外面明明套了 with connection，重查数据库却发现批次和明细都还在。遇到这种情况，先沿调用链检查谁执行了提交，尤其是公共函数里的第二层 with connection。
下面是可执行的缩小案例，不对应某个生产事故。适用于 Python 3.12 及以上的标准库 sqlite3；示例显式选择 autocommit=False，不依赖随版本变化的默认模式。本次在 Windows、Python 3.14.6、SQLite 3.50.4 上验证。
现象：异常传播正常，原子性仍然失效
假设一批导入有两个动作：写入 batch 标记，再写入 detail 标记。业务要求全部成功或全部撤销。公共写入函数为了“保证事务”，又包了一层连接上下文：
import sqlite3
from contextlib import closing
with closing(sqlite3.connect(":memory:", autocommit=False)) as connection:
connection.execute("CREATE TABLE event (name TEXT PRIMARY KEY)")
connection.commit()
try:
with connection:
connection.execute("INSERT INTO event VALUES (?)", ("batch",))
with connection:
connection.execute("INSERT INTO event VALUES (?)", ("detail",))
raise RuntimeError("批次校验失败")
except RuntimeError as exc:
print(str(exc))
print(connection.execute("SELECT name FROM event ORDER BY name").fetchall())
运行结果是“批次校验失败”，随后输出 [('batch',), ('detail',)]。这不是异常被吃掉：外层确实收到异常并执行了回滚，只是原来的两条写入已经被内层正常退出时提交了。
根因：连接上下文没有嵌套层级
with connection 的退出动作针对连接当前的事务，不会因 Python 缩进自动创建保存点。内层正常退出时提交的是连接事务，其中也包含外层先前写入的 batch。这与 Django 的嵌套 atomic() 语义不同，不能直接照搬使用经验。
在本例的 autocommit=False 模式中，提交后驱动会重新开启事务。因此，只看 connection.in_transaction 一直为真，不能证明最初的事务还没有提交。用 autocommit=True 替换也不是修复：此时连接上下文不会替你提交或回滚，写入的原子性需要另行管理。连接上下文的行为见 Python sqlite3 官方说明。
排查时应把事务边界与业务动作放在同一条时间线上：入口开始批次、写主记录、调用公共函数、公共函数退出、入口失败。检查公共函数是否调用 commit()、rollback()、另开连接或使用连接上下文。不要在生产请求中直接记录 SQL 全文或参数；它们可能含业务敏感数据。
修复：由业务入口统一拥有事务
这个场景要求整批原子性，公共函数只执行参数化语句，让异常向入口传播即可。不需要为每个函数都添加事务管理器。下面的完整文件同时包含故障实现、修复实现和八项测试。保存为 test_transaction_owner.py，执行 python -m unittest -v test_transaction_owner。
import sqlite3
import unittest
from collections.abc import Callable
def write_detail_broken(connection: sqlite3.Connection, name: str) -> None:
"""保留故障实现，用于验证提前提交现象。"""
with connection:
connection.execute("INSERT INTO event VALUES (?)", (name,))
def write_detail(connection: sqlite3.Connection, name: str) -> None:
"""加入调用方事务；提交和回滚由业务入口负责。"""
connection.execute("INSERT INTO event VALUES (?)", (name,))
def import_batch(
connection: sqlite3.Connection,
writer: Callable[[sqlite3.Connection, str], None],
*,
should_fail: bool = False,
detail_name: str = "detail",
) -> None:
"""演示批次入口；调用前不应存在其他待提交的业务工作。"""
with connection:
connection.execute("INSERT INTO event VALUES (?)", ("batch",))
writer(connection, detail_name)
if should_fail:
raise RuntimeError("批次校验失败")
class TransactionOwnerTests(unittest.TestCase):
def setUp(self) -> None:
self.connection = sqlite3.connect(":memory:", autocommit=False)
self.addCleanup(self.connection.close)
self.connection.execute("CREATE TABLE event (name TEXT PRIMARY KEY)")
self.connection.commit()
def names(self) -> list[str]:
return [
row[0]
for row in self.connection.execute("SELECT name FROM event ORDER BY name")
]
def test_broken_writer_commits_before_outer_failure(self) -> None:
with self.assertRaisesRegex(RuntimeError, "批次校验失败"):
import_batch(self.connection, write_detail_broken, should_fail=True)
self.assertEqual(self.names(), ["batch", "detail"])
def test_outer_failure_rolls_back_every_write(self) -> None:
with self.assertRaisesRegex(RuntimeError, "批次校验失败"):
import_batch(self.connection, write_detail, should_fail=True)
self.assertEqual(self.names(), [])
def test_success_survives_later_rollback(self) -> None:
import_batch(self.connection, write_detail)
self.connection.rollback()
self.assertEqual(self.names(), ["batch", "detail"])
def test_duplicate_detail_rolls_back_batch(self) -> None:
self.connection.execute("INSERT INTO event VALUES (?)", ("detail",))
self.connection.commit()
with self.assertRaises(sqlite3.IntegrityError):
import_batch(self.connection, write_detail)
self.assertEqual(self.names(), ["detail"])
def test_writer_does_not_commit_callers_work(self) -> None:
self.connection.execute("INSERT INTO event VALUES (?)", ("before",))
write_detail(self.connection, "detail")
self.connection.rollback()
self.assertEqual(self.names(), [])
def test_value_is_bound_as_data(self) -> None:
name = "x'); DROP TABLE event; --"
import_batch(self.connection, write_detail, detail_name=name)
self.assertEqual(self.names(), ["batch", name])
def test_released_savepoint_still_rolls_back_with_parent(self) -> None:
with self.assertRaisesRegex(RuntimeError, "批次校验失败"):
with self.connection:
self.connection.execute("INSERT INTO event VALUES (?)", ("batch",))
self.connection.execute("SAVEPOINT optional_detail")
write_detail(self.connection, "detail")
self.connection.execute("RELEASE SAVEPOINT optional_detail")
raise RuntimeError("批次校验失败")
self.assertEqual(self.names(), [])
def test_local_rollback_preserves_parent_work(self) -> None:
with self.connection:
self.connection.execute("INSERT INTO event VALUES (?)", ("batch",))
self.connection.execute("SAVEPOINT optional_detail")
write_detail(self.connection, "detail")
self.connection.execute("ROLLBACK TO SAVEPOINT optional_detail")
self.connection.execute("RELEASE SAVEPOINT optional_detail")
self.assertEqual(self.names(), ["batch"])
if __name__ == "__main__":
unittest.main(verbosity=2)
这里让 writer 可替换，是为了用相同入口对比两种事务行为。测试不仅检查成功写入，也检查入口失败时是否一条不留、唯一约束错误是否传播、公共函数是否擅自提交调用方的数据。SQL 参数化不负责业务格式校验；生产入口仍应按实际字段规则校验名称的类型、长度和允许值。
closing() 与事务上下文也承担不同职责：前者关闭连接，后者处理事务。本测试使用 addCleanup() 关闭每条独立连接，避免把资源释放误认为提交策略。
如果业务确实需要局部回滚
例如可选明细失败后允许保留主记录，可以在已经存在的外层事务内建立 SAVEPOINT，按业务约定执行 ROLLBACK TO，再 RELEASE。上面最后两项测试分别证明：释放内层保存点后，外层仍能撤销全部；撤回到保存点后，外层早先的写入仍可提交。
ROLLBACK TO 后保存点还在；RELEASE 用于结束该保存点。若保存点本身就是最外层事务，释放它会提交，因此不能省略对外层事务是否存在的判断。保存点规则见 SQLite 官方文档。示例名字是固定的内部标识符，不接受用户输入；业务值仍通过占位符绑定。生产封装还要处理嵌套命名、释放失败及异常链，不应把这两段测试直接当成通用事务库。
验证结果与上线检查
本次执行上面的八项测试全部通过。随后把修复函数临时改回嵌套 with connection，外层失败全撤销、公共函数不提交上游工作两项回归断言失败；两项保存点测试也因提前提交清除了保存点而报错。恢复后八项再次通过。这证明测试能识别原来的提交缺陷及其对保存点的影响。复现段的残留两行也已实际运行确认。
验证使用独立的内存数据库，检查的是事务逻辑；没有验证生产文件持久化、多进程锁竞争、断电恢复、ORM 或真实导入服务。环境没有 Ruff，未运行格式化和静态检查，也未为文章安装依赖。
上线前先梳理所有写入入口。移除公共函数里的提交行为后，独立调用它的脚本也必须显式拥有事务，否则可能在关闭连接时丢弃尚未提交的工作。不要只改公共函数而漏掉调用方。请求结束前应保持“成功提交、失败回滚”的单一出口，并增加与实际业务表结构对应的失败注入用例。
如果旧代码已留下部分批次，先暂停有问题的入口，以业务批次号核对实际明细与处理状态，再制定补偿方案。不要仅因遇到这类问题就批量删除记录。回退时也应整体回退入口与公共函数的事务约定，避免新旧提交策略混用。
总结
with connection 能处理连接当前事务的退出，却不代表可随意嵌套。整批成功或失败的事务应由业务入口统一管理，底层函数只执行语句；需要局部撤销时再引入明确的保存点。回归测试要在中间步骤之后主动制造失败，并检查最终数据，才能发现“异常抛出了，但数据已经提交”的问题。
Discussion
评论