Python实现MySQL错误回滚机制的实战代码

pip install pymysql

import pymysql
import sys

# 数据库配置（请根据你的实际情况修改）
DB_CONFIG = {
    'host': 'localhost',
    'user': 'root',
    'password': 'your_password',
    'database': 'test_db',
    'charset': 'utf8mb4'
}

def setup_database(cursor):
    """初始化测试表和数据"""
    try:
        cursor.execute("DROP TABLE IF EXISTS accounts")
        cursor.execute("""
            CREATE TABLE accounts (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(50),
                balance DECIMAL(10, 2)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """)
        # 插入初始数据：Alice有1000元，Bob有500元
        cursor.execute("INSERT INTO accounts (name, balance) VALUES ('Alice', 1000.00)")
        cursor.execute("INSERT INTO accounts (name, balance) VALUES ('Bob', 500.00)")
        print("✅ 数据库初始化完成：Alice=1000, Bob=500")
    except Exception as e:
        print(f"❌ 初始化失败: {e}")

def transfer_money_with_rollback(from_user, to_user, amount):
    """
    模拟转账业务，并在发生错误时回滚
    """
    connection = None
    try:
        # 1. 建立连接
        connection = pymysql.connect(**DB_CONFIG)
        
        # 2. 关键步骤：关闭自动提交，开启事务
        connection.autocommit(False)
        
        with connection.cursor() as cursor:
            # --- 步骤一：扣款 ---
            print(f"\n💰 正在从 {from_user} 扣除 {amount} 元...")
            sql_deduct = "UPDATE accounts SET balance = balance - %s WHERE name = %s"
            cursor.execute(sql_deduct, (amount, from_user))
            
            # 模拟查询扣款后的余额（仅为了演示，实际业务中可能不需要）
            cursor.execute("SELECT balance FROM accounts WHERE name = %s", (from_user,))
            result = cursor.fetchone()
            print(f"   👉 扣款后查询 {from_user} 余额: {result[0]} (此时数据在内存/Redo Log中，未永久落盘)")

            # --- 步骤二：模拟突发异常 ---
            # 比如：此时服务器断电、网络中断、或者代码逻辑错误
            print("⚠️  模拟系统崩溃：准备加款时发生除零错误！")
            error_simulation = 1 / 0  # 故意制造一个异常
            
            # --- 步骤三：加款（正常情况下会执行，但上面报错了就不会走到这） ---
            sql_add = "UPDATE accounts SET balance = balance + %s WHERE name = %s"
            cursor.execute(sql_add, (amount, to_user))
            
            # 3. 如果一切顺利，提交事务
            connection.commit()
            print("✅ 转账成功，事务已提交！")

    except Exception as e:
        print(f"\n❌ 发生严重错误: {e}")
        if connection:
            # 4. 核心：发生任何异常，回滚所有操作
            print("🔄 正在执行回滚操作 (ROLLBACK)...")
            connection.rollback()
            print("🛡️ 回滚成功！数据已恢复到事务开始前的状态。")
            
    finally:
        if connection:
            # 5. 恢复自动提交模式并关闭连接
            connection.autocommit(True)
            connection.close()

def check_final_balance():
    """检查最终结果"""
    conn = pymysql.connect(**DB_CONFIG)
    with conn.cursor() as cursor:
        cursor.execute("SELECT name, balance FROM accounts")
        results = cursor.fetchall()
        print("\n----- 最终账户余额 -----")
        for row in results:
            print(f"用户: {row[0]}, 余额: {row[1]}")
        print("------------------------")
        # 验证结果
        alice_balance = results[0][1] if results[0][0] == 'Alice' else results[1][1]
        bob_balance = results[0][1] if results[0][0] == 'Bob' else results[1][1]
        assert alice_balance == 1000, f"Alice余额错误！期望1000，实际{alice_balance}"
        assert bob_balance == 500, f"Bob余额错误！期望500，实际{bob_balance}"
        print("🎉 验证通过：数据一致，回滚生效！")
    conn.close()

if __name__ == "__main__":
    # 初始化
    conn_init = pymysql.connect(**DB_CONFIG)
    with conn_init.cursor() as cur:
        setup_database(cur)
        conn_init.commit()
    conn_init.close()

    # 执行带回滚的转账
    transfer_money_with_rollback('Alice', 'Bob', 200)

    # 检查最终数据是否正确回滚
    check_final_balance()