Python 数据库事务：commit 与 rollback 操作

在处理数据库操作时，确保数据的完整性和一致性至关重要。事务管理是解决这一问题的核心机制，它将一系列操作捆绑在一起，要么全部成功，要么全部失败。本指南将详细介绍如何在 Python 中使用 commit（提交）和 rollback（回滚）来控制数据库事务。

1. 理解事务的核心概念

想象一下银行转账过程：从账户 A 扣除 100 元，向账户 B 增加 100 元。这两个操作必须是一个不可分割的整体。如果钱从 A 扣除了但未转入 B，数据就会出错。

事务包含四个关键属性（ACID），这里重点关注最基础的两个：

• 原子性：事务中的所有操作要么全部完成，要么全部不完成，不会结束在中间某个环节。
• 一致性：事务开始前和结束后，数据库的完整性约束没有被破坏。

在 Python 的数据库操作中，默认情况下往往处于“自动提交”模式，即每执行一条 SQL 语句（如 INSERT、UPDATE），更改立即生效。为了使用事务，我们需要关闭自动提交，手动控制何时保存更改。

2. 准备工作：建立数据库连接

本指南使用 Python 内置的 sqlite3 库进行演示，逻辑同样适用于 MySQL 或 PostgreSQL 等其他数据库。

创建一个名为 transaction_demo.py 的文件。输入以下代码来初始化一个简单的数据库表：

import sqlite3

# 连接到数据库（如果不存在会自动创建）
conn = sqlite3.connect('bank.db')

# 创建一个游标对象，用于执行 SQL 命令
cursor = conn.cursor()

# 创建账户表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS accounts (
        id INTEGER PRIMARY KEY,
        name TEXT,
        balance INTEGER
    )
''')

# 插入初始数据
cursor.execute("INSERT OR IGNORE INTO accounts (id, name, balance) VALUES (1, 'Alice', 1000)")
cursor.execute("INSERT OR IGNORE INTO accounts (id, name, balance) VALUES (2, 'Bob', 1000)")

# 这里先提交一次初始结构
conn.commit()
print("数据库初始化完成。")

运行该脚本以确保环境就绪。

3. 正确提交：Commit 操作

当确认一系列操作都正确无误后，必须显式调用 commit() 方法，此时更改才会永久写入数据库。

修改脚本，执行一笔正常的转账操作：Alice 转 100 元给 Bob。

1. 开启一个事务概念块（在 SQLite 中，不使用 isolation_level 参数时，commit 前的操作都在事务中）。
2. 执行扣除 Alice 余额的 SQL。
3. 执行增加 Bob 余额的 SQL。
4. 调用 conn.commit() 保存更改。

代码示例如下：

# 假设已建立 conn 和 cursor

try:
    # 1. 扣除 Alice 100元
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE name = 'Alice'")

    # 2. 给 Bob 增加 100元
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE name = 'Bob'")

    # 3. 检查结果（仅在内存中，未写入磁盘）
    print("操作执行完毕，准备提交...")

    # 4. 提交事务
    conn.commit()
    print("事务提交成功！更改已保存。")

except Exception as e:
    print(f"发生错误: {e}")

如果不执行第 4 步的 conn.commit()，或者程序在到达第 4 步前崩溃，Alice 和 Bob 的余额将保持原样，仿佛什么都没发生过。

4. 错误处理：Rollback 操作

如果在事务执行过程中发生了任何错误（如余额不足、SQL 语法错误、网络断开），应该调用 rollback() 方法。这将撤销自上次提交以来的所有更改，将数据库恢复到事务开始前的状态。

模拟一个异常场景：尝试转账金额超过余额，或者手动制造一个错误。

1. 执行扣除 Alice 的操作。
2. 模拟一个错误（例如手动抛出异常）。
3. 捕获异常。
4. 调用 conn.rollback()。

代码示例如下：

try:
    # 扣除 Alice 5000元（假设余额不足或其他逻辑错误）
    cursor.execute("UPDATE accounts SET balance = balance - 5000 WHERE name = 'Alice'")

    # 模拟一个运行时错误
    raise Exception("银行系统维护中，交易取消")

    # 下面的代码不会执行
    cursor.execute("UPDATE accounts SET balance = balance + 5000 WHERE name = 'Bob'")

    conn.commit()

except Exception as e:
    print(f"交易失败: {e}")
    # 核心步骤：回滚所有更改
    conn.rollback()
    print("已执行回滚，数据未受影响。")

# 验证数据：查询余额
cursor.execute("SELECT name, balance FROM accounts")
print("当前余额:", cursor.fetchall())

执行上述代码后，你会发现 Alice 的余额并没有减少。rollback() 确保了数据的“要么全有，要么全无”。

5. 事务流程逻辑

为了更直观地理解 commit 和 rollback 的决策过程，请参考以下流程图：

注意观察图表中的逻辑路径：只有当所有步骤都标为“是”时，才会走向 commit 分支；任何偏差都会导致流向 rollback。

6. 使用上下文管理器（with 语句）简化代码

Python 提供了更优雅的方式来处理事务，利用上下文管理器可以自动处理提交和回滚。如果在 with 块中发生异常，事务会自动回滚；如果没有异常，则自动提交。

使用 connection 对象作为上下文管理器：

# 创建连接
conn = sqlite3.connect('bank.db')

# 使用 with 语句
with conn:
    # 在这个缩进块内的所有操作都属于同一个事务
    cursor = conn.cursor()

    # 这里的 cursor.execute 需要重新获取 cursor，或者使用 conn.execute
    cursor.execute("UPDATE accounts SET balance = balance - 200 WHERE name = 'Alice'")

    # 模拟错误以测试自动回滚
    # raise ValueError("测试错误")

    cursor.execute("UPDATE accounts SET balance = balance + 200 WHERE name = 'Bob'")

    # 如果代码运行到这里没有报错，Python 会自动调用 conn.commit()
    print("自动提交已触发。")

# 如果上面的代码块中发生了错误，Python 会自动调用 conn.rollback()
print("事务处理结束。")

这种方式省去了显式的 try...except...finally 样板代码，降低了人为忘记提交或回滚的风险。

7. 实战场景总结表

下表总结了在不同场景下应采取的操作：

8. 完整的转账函数示例

将上述知识整合为一个完整的、可复用的转账函数。该函数包含逻辑检查（余额是否充足）和事务保护。

def transfer_money(db_conn, from_user, to_user, amount):
    """
    执行转账操作，包含事务控制。
    """
    try:
        cursor = db_conn.cursor()

        # 1. 检查转出账户余额是否充足
        cursor.execute("SELECT balance FROM accounts WHERE name = ?", (from_user,))
        result = cursor.fetchone()

        if result is None:
            raise ValueError(f"用户 {from_user} 不存在")

        current_balance = result[0]
        if current_balance < amount:
            raise ValueError(f"余额不足！当前余额: {current_balance}, 转账金额: {amount}")

        # 2. 执行转账（在事务中）
        # 扣除
        cursor.execute("UPDATE accounts SET balance = balance - ? WHERE name = ?", (amount, from_user))
        # 增加
        cursor.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?", (amount, to_user))

        # 3. 提交事务
        db_conn.commit()
        print(f"转账成功: {from_user} 向 {to_user} 转账 {amount} 元。")
        return True

    except Exception as e:
        # 4. 发生任何错误，回滚事务
        db_conn.rollback()
        print(f"转账失败: {e}")
        return False

# 测试函数
conn = sqlite3.connect('bank.db')
# 第一次测试：成功
transfer_money(conn, 'Alice', 'Bob', 300)
# 第二次测试：失败（余额不足）
transfer_money(conn, 'Alice', 'Bob', 10000)

conn.close()

验证结果：你可以查询数据库表，确认第一次转账的余额变化已保存，而第二次转账的失败没有破坏第一次的数据一致性。