Python sqlite3 事务模式与连接上下文管理

Python 的 sqlite3 事务代码最容易混淆的不是 SQL，而是三个相似但职责不同的入口：autocommit、isolation_level 和 with con。直接给出选择结论：Python 3.12 及以上的新代码，普通请求级写入优先显式设置 autocommit=False，再用连接上下文管理器统一提交或回滚；每条语句都应独立提交时用 autocommit=True；只有维护旧代码时，才继续依赖 LEGACY_TRANSACTION_CONTROL 与 isolation_level。

先按工作负载决定事务边界

我在一个本地任务队列里遇到过典型问题：一次请求要写入任务、扣减配额、追加审计记录，这三条 SQL 必须一起成功；另一个清理任务却只执行一条 DELETE，失败后重试即可。两类负载如果套用同一种事务模板，要么事务过长，要么原子性不足。

这里的“模式”不是性能等级。SQLite 允许多个连接同时读，但同一时刻只允许一个写事务。事务范围越大，原子性越清晰，写锁持续时间也可能越长；范围越小，并发等待更少，跨语句一致性则要由业务重新设计。

三种事务控制模式怎么选

从 Python 3.12 开始，官方推荐通过 Connection.autocommit 控制事务。它有三个有意义的取值，而不是简单的“开”和“关”。

autocommit=False：PEP 249 风格的请求级事务

这个模式下，sqlite3 会保证总有一个事务处于打开状态。连接建立后会用 BEGIN DEFERRED 打开事务；调用 commit() 或 rollback() 结束当前事务后，又会立即打开一个新事务。因此它适合“一个连接承载一段明确业务工作”的代码。

import sqlite3
from contextlib import closing


def create_order(db_path: str, user_id: int, amount: int) -> int:
    # closing 负责连接释放，autocommit=False 负责 PEP 249 事务语义。
    with closing(sqlite3.connect(db_path, autocommit=False)) as con:
        try:
            with con:
                # 三条写入属于同一个业务工作单元。
                cursor = con.execute(
                    "INSERT INTO orders(user_id, amount) VALUES (?, ?)",
                    (user_id, amount),
                )
                order_id = cursor.lastrowid
                con.execute(
                    "UPDATE quota SET remaining = remaining - ? WHERE user_id = ?",
                    (amount, user_id),
                )
                con.execute(
                    "INSERT INTO audit(order_id, action) VALUES (?, ?)",
                    (order_id, "created"),
                )
            return int(order_id)
        except sqlite3.Error:
            # with con 已处理回滚；这里保留异常供上层记录或重试。
            raise

如果 with con 内没有未捕获异常，退出时提交；如果 SQL 或业务代码抛出异常，退出时回滚，然后异常继续向外传播。不要在块内捕获异常后悄悄返回，否则上下文管理器会看到“正常退出”并提交前面已经完成的语句。

autocommit=True：每条独立语句交给 SQLite

设置为 True 后，底层 SQLite 处于自动提交模式。每条独立语句会在自己的隐式事务里完成，Python 的 commit() 和 rollback() 此时没有效果。适合幂等、彼此无关的短操作，不适合需要三条 SQL 要么全部成功、要么全部撤销的业务。

import sqlite3
from contextlib import closing


def delete_expired_jobs(db_path: str, deadline: str) -> int:
    # 单条删除就是完整工作单元，不额外维持 Python 事务。
    with closing(sqlite3.connect(db_path, autocommit=True)) as con:
        cursor = con.execute(
            "DELETE FROM jobs WHERE expires_at 

LEGACY_TRANSACTION_CONTROL：旧代码的兼容入口

当前默认值仍是 sqlite3.LEGACY_TRANSACTION_CONTROL，但官方文档已经说明未来默认值会改为 False。在兼容模式中，isolation_level 才决定隐式执行哪种 BEGIN：默认 "DEFERRED"，也可设置 "IMMEDIATE"、"EXCLUSIVE"，或者用 None 禁用隐式开启事务。

迁移时不要同时写 autocommit=False 和 isolation_level="IMMEDIATE"，然后期待立即取得写锁。只要 autocommit 不是旧兼容常量，isolation_level 就没有作用。应先明确要保留旧语义，还是切换到新的事务控制方式。

with con 管事务，不负责关闭连接

连接对象作为上下文管理器时，只处理“块结束后怎样收尾当前事务”。它既不会在进入块时必然开启新事务，也不会在退出块时调用 close()。Python 3.13 起，如果 Connection 被删除前没有关闭，还会发出 ResourceWarning，因此资源生命周期最好显式写出来。

最清晰的组合是外层 contextlib.closing() 负责连接，内层 with con 负责事务。若 autocommit=True，上下文管理器在退出时不会做事务处理；若 autocommit=False，提交或回滚后会立刻开启下一事务，随后外层关闭连接时会回滚这个尚无改动的新事务。

import sqlite3
from contextlib import closing


def rename_user(db_path: str, user_id: int, new_name: str) -> None:
    # 外层只管理连接生命周期。
    with closing(sqlite3.connect(db_path, autocommit=False)) as con:
        # 内层只管理这一段业务事务。
        with con:
            con.execute(
                "UPDATE users SET name = ? WHERE id = ?",
                (new_name, user_id),
            )
            if con.total_changes != 1:
                # 抛出异常让上下文管理器回滚，而不是提交空更新。
                raise LookupError(f"user not found: {user_id}")

另一个常见误区是嵌套两个 with con，把内层当成子事务。连接上下文管理器不会创建 SAVEPOINT；内层正常退出可能提交整个当前事务，破坏外层原子性。需要嵌套工作单元时，应显式使用 SAVEPOINT。

写锁竞争与嵌套工作单元怎么处理

BEGIN DEFERRED 会推迟到首次访问数据库才真正开始事务。如果先读后写，另一个连接可能已经取得写事务，当前连接升级时就会遇到 SQLITE_BUSY。确定马上要写、并且希望在业务开始前就暴露竞争时，可以使用 BEGIN IMMEDIATE。

为了避免与 Python 自动事务控制打架，可以在 autocommit=True 下用 SQL 明确管理这段事务。注意此时必须执行 SQL COMMIT/ROLLBACK，不能依赖无效的 con.commit() 和 con.rollback()。

import sqlite3
from contextlib import closing


def claim_job(db_path: str, worker: str) -> int | None:
    # timeout 控制锁冲突时最多等待多久再抛出 OperationalError。
    with closing(sqlite3.connect(db_path, timeout=3.0, autocommit=True)) as con:
        try:
            con.execute("BEGIN IMMEDIATE")  # 先取得写事务，避免读完后升级失败。
            row = con.execute(
                "SELECT id FROM jobs WHERE status = 'pending' ORDER BY id LIMIT 1"
            ).fetchone()
            if row is None:
                con.execute("COMMIT")  # 没有任务也结束显式事务。
                return None

            job_id = int(row[0])
            con.execute(
                "UPDATE jobs SET status = 'running', worker = ? WHERE id = ?",
                (worker, job_id),
            )
            con.execute("COMMIT")  # 自动提交模式下用 SQL 结束显式事务。
            return job_id
        except Exception:
            if con.in_transaction:
                # 只在底层事务仍打开时回滚，避免覆盖原始异常。
                con.execute("ROLLBACK")
            raise

SQLite 的 EXCLUSIVE 与 IMMEDIATE 都会立即开始写事务；在 WAL 模式下两者相同，在其他日志模式下，EXCLUSIVE 还会阻止其他连接读取。除非确实需要这个读阻塞边界，否则普通写入通常不必选择 EXCLUSIVE。

需要部分回滚时，用 SAVEPOINT 表达嵌套工作单元：

def reserve_stock(con: sqlite3.Connection, sku: str, quantity: int) -> None:
    con.execute("SAVEPOINT reserve_stock")  # 在外层事务内建立可独立撤销的保存点。
    try:
        cursor = con.execute(
            "UPDATE stock SET available = available - ? "
            "WHERE sku = ? AND available >= ?",
            (quantity, sku, quantity),
        )
        if cursor.rowcount != 1:
            raise ValueError("insufficient stock")
    except Exception:
        con.execute("ROLLBACK TO SAVEPOINT reserve_stock")  # 撤销保存点后的修改。
        con.execute("RELEASE SAVEPOINT reserve_stock")  # 释放保存点名称。
        raise
    else:
        con.execute("RELEASE SAVEPOINT reserve_stock")  # 合并到外层事务，尚未最终提交。

迁移风险与落地清单

不要依赖未写明的默认值。 当前 connect() 的 autocommit 默认仍是 LEGACY_TRANSACTION_CONTROL，未来会改成 False。新代码显式传入目标值，升级时就不会因为默认切换改变事务时机。

谨慎使用 executescript()。 在旧兼容事务控制下，executescript() 会在执行脚本前隐式提交待处理事务，不受 isolation_level 影响。迁移脚本若要求整体原子性，应在脚本中明确写事务语句，并单独测试失败分支。

把锁等待当成正常分支。 多进程或多连接写同一个数据库时，OperationalError: database is locked 不是只靠扩大 timeout 就能根治。先缩短事务、避免事务内做网络请求，再按业务幂等性设计有限次数重试。

连接不要跨线程随意共享。 默认 check_same_thread=True 会阻止连接被创建线程之外的线程使用。设置为 False 不等于自动获得安全的并发写入，应用仍要序列化同一连接上的写操作。

最终落地时可以逐项核对：连接是否显式设置 autocommit；一次业务操作包含哪些 SQL；异常是否能逃出 with con 触发回滚；连接是否由 closing 或明确的 close() 释放；是否存在嵌套 with con；锁冲突是否有超时与有限重试；旧代码中的 isolation_level 是否仍处于兼容模式。把这些边界写进数据访问层，比在每个调用点猜测当前事务状态可靠得多。

几个容易继续追问的问题

with sqlite3.connect(...) as con 会关闭连接吗

不会。它只在退出时提交或回滚打开的事务。需要自动关闭时，再包一层 contextlib.closing()，或者在 finally 中调用 close()。

autocommit=False 为什么提交后仍显示在事务中

这是 PEP 249 模式的设计：commit() 结束当前事务后，sqlite3 会立即隐式打开新事务。要观察底层 SQLite 是否处于事务中，可读取 in_transaction，但不要把它当成业务事务边界的唯一设计依据。

isolation_level="IMMEDIATE" 为什么不生效

先检查 autocommit。只有它等于 sqlite3.LEGACY_TRANSACTION_CONTROL 时，isolation_level 才控制隐式 BEGIN 类型；在 autocommit=False 或 True 下，该属性不参与事务控制。

官方资料：https://docs.python.org/3/library/sqlite3.html#transaction-control、https://docs.python.org/3/library/sqlite3.html#how-to-use-the-connection-context-manager、https://www.sqlite.org/lang_transaction.html。