Python 使用 MySQL 数据库进行事务处理完整示例

事务Transaction是数据库操作的最小逻辑单元遵循ACID 原则原子性Atomicity要么全部执行成功要么全部失败回滚一致性Consistency执行前后数据库状态保持一致隔离性Isolation多个事务互不干扰持久性Durability提交后数据永久保存Python操作MySQL时默认是自动提交autocommit模式执行单条SQL会立即生效而事务处理需要先关闭自动提交手动控制提交/回滚。二、完整事务处理代码示例以「转账场景」为例经典的事务应用场景A账户扣钱、B账户加钱必须同时成功/失败1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859importmysql.connectorfrommysql.connectorimportErrordeftransfer_money(from_id, to_id, amount):模拟转账的事务处理:param from_id: 转出账户ID:param to_id: 转入账户ID:param amount: 转账金额connectionNonetry:# 1. 建立数据库连接connectionmysql.connector.connect(hostlocalhost,userroot,password123456,databasetest_db)# 2. 关闭自动提交开启手动事务控制connection.autocommitFalsecursorconnection.cursor()# 3. 执行事务内的多个SQL操作# 步骤1扣除转出账户金额deduct_sqlUPDATE account SET balance balance - %s WHERE id %scursor.execute(deduct_sql, (amount, from_id))# 步骤2增加转入账户金额add_sqlUPDATE account SET balance balance %s WHERE id %scursor.execute(add_sql, (amount, to_id))# 模拟异常可取消注释测试回滚效果# raise Error(模拟转账异常触发回滚)# 4. 所有操作执行成功提交事务connection.commit()print(转账成功事务已提交)exceptError as e:# 5. 发生异常回滚事务撤销所有已执行的SQL操作ifconnection:connection.rollback()print(f转账失败事务已回滚错误信息{e})finally:# 6. 释放资源关闭游标和连接ifconnectionandconnection.is_connected():cursor.close()# 恢复自动提交可选不影响但规范connection.autocommitTrueconnection.close()print(数据库连接已关闭)# 测试前准备 # 先在MySQL中创建测试表和数据# CREATE DATABASE IF NOT EXISTS test_db;# USE test_db;# CREATE TABLE IF NOT EXISTS account (# id INT PRIMARY KEY,# name VARCHAR(50),# balance DECIMAL(10,2)# );# INSERT INTO account (id, name, balance) VALUES (1, 张三, 1000.00), (2, 李四, 500.00);# 执行转账测试 # 测试正常转账张三给李四转200元transfer_money(from_id1, to_id2, amount200.00)三、关键操作解释关闭自动提交connection.autocommit False是开启事务的核心关闭后执行的SQL不会立即生效需手动commit()。事务提交connection.commit()仅当所有SQL执行无异常时调用会将事务内所有操作永久写入数据库。事务回滚connection.rollback()仅在捕获到异常时调用会撤销事务内所有已执行的SQL操作恢复到事务开始前的状态。异常捕获必须用try-except包裹所有事务内的操作确保任何错误都能触发回滚避免数据不一致。四、拓展场景批量操作事务如果需要批量插入/更新多条数据事务同样适用示例如下1234567891011121314151617181920212223242526272829defbatch_insert_users(users):批量插入用户使用事务保证全部成功/失败connectionNonetry:connectionmysql.connector.connect(hostlocalhost,userroot,password123456,databasetest_db)connection.autocommitFalsecursorconnection.cursor()insert_sqlINSERT INTO user (name, age) VALUES (%s, %s)# 批量执行SQL效率更高cursor.executemany(insert_sql, users)connection.commit()print(f批量插入 {cursor.rowcount} 条数据成功)exceptError as e:ifconnection:connection.rollback()print(f批量插入失败事务回滚{e})finally:ifconnectionandconnection.is_connected():cursor.close()connection.autocommitTrueconnection.close()# 测试批量插入user_list[(王五,30), (赵六,28), (孙七,35)]batch_insert_users(user_list)到此这篇关于Python 使用 MySQL 数据库进行事务处理步骤的文章就介绍到这了