当前位置：首页 > news >正文

Python MySQL 错误回滚实战代码

news 2026/5/12 0:53:31

这个例子模拟了一个经典的“转账”场景：A 给 B 转钱，如果在扣款后、收款前系统发生错误（比如断电、代码异常），必须让数据回到转账前的状态，保证钱不凭空消失。

环境准备：
你需要安装pymysql库：

pipinstallpymysql

代码实现：

importpymysqlimportsys# 数据库配置（请根据你的实际情况修改）DB_CONFIG={'host':'localhost','user':'root','password':'your_password','database':'test_db','charset':'utf8mb4'}defsetup_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")exceptExceptionase:print(f"❌ 初始化失败:{e}")deftransfer_money_with_rollback(from_user,to_user,amount):""" 模拟转账业务，并在发生错误时回滚 """connection=Nonetry:# 1. 建立连接connection=pymysql.connect(**DB_CONFIG)# 2. 关键步骤：关闭自动提交，开启事务connection.autocommit(False)withconnection.cursor()ascursor:# --- 步骤一：扣款 ---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("✅ 转账成功，事务已提交！")exceptExceptionase:print(f"\n❌ 发生严重错误:{e}")ifconnection:# 4. 核心：发生任何异常，回滚所有操作print("🔄 正在执行回滚操作 (ROLLBACK)...")connection.rollback()print("🛡️ 回滚成功！数据已恢复到事务开始前的状态。")finally:ifconnection:# 5. 恢复自动提交模式并关闭连接connection.autocommit(True)connection.close()defcheck_final_balance():"""检查最终结果"""conn=pymysql.connect(**DB_CONFIG)withconn.cursor()ascursor:cursor.execute("SELECT name, balance FROM accounts")results=cursor.fetchall()print("\n----- 最终账户余额 -----")forrowinresults:print(f"用户:{row[0]}, 余额:{row[1]}")print("------------------------")# 验证结果alice_balance=results[0][1]ifresults[0][0]=='Alice'elseresults[1][1]bob_balance=results[0][1]ifresults[0][0]=='Bob'elseresults[1][1]assertalice_balance==1000,f"Alice余额错误！期望1000，实际{alice_balance}"assertbob_balance==500,f"Bob余额错误！期望500，实际{bob_balance}"print("🎉 验证通过：数据一致，回滚生效！")conn.close()if__name__=="__main__":# 初始化conn_init=pymysql.connect(**DB_CONFIG)withconn_init.cursor()ascur:setup_database(cur)conn_init.commit()conn_init.close()# 执行带回滚的转账transfer_money_with_rollback('Alice','Bob',200)# 检查最终数据是否正确回滚check_final_balance()

代码讲解重点（写进博客里）：

connection.autocommit(False)：这是事务的开关。默认情况下 MySQL 是自动提交的（每句 SQL 都是一个事务），关掉它才能把多步操作打包成一个整体。
try...except...结构：业务逻辑必须放在try里。
connection.rollback()：这是“后悔药”。一旦进入except块，调用此方法会撤销从autocommit(False)之后的所有未提交更改。
connection.commit()：这是“确认键”。只有执行了这个，数据才真正写入磁盘（配合 Redo Log 和 Binlog）。
Engine=InnoDB：注意建表时指定了引擎为 InnoDB。如果是 MyISAM 引擎，它不支持事务，rollback会失效！这是面试常考点。