Python 操作 MySQL 事务：从入门到避坑

在实际开发中，单条 SQL 往往不够用。转账、订单处理、库存扣减……这些场景要求多条 SQL要么全成功，要么全失败。这就是事务存在的意义。

本文用 Python 实战讲解如何正确使用 MySQL 事务，覆盖pymysql、mysql-connector-python和SQLAlchemy三种主流方式。

一、先搞懂事务的核心：ACID

一句话：事务就是保证数据不出乱子的机制。

二、方式一：pymysql（最常用）

2.1 基本用法

importpymysql conn=pymysql.connect(host='localhost',user='root',password='your_password',database='test_db',charset='utf8mb4')try:withconn.cursor()ascursor:# 开启事务（默认就是手动提交模式）sql1="UPDATE accounts SET balance = balance - 100 WHERE user_id = 1"sql2="UPDATE accounts SET balance = balance + 100 WHERE user_id = 2"cursor.execute(sql1)cursor.execute(sql2)# 全部成功，提交conn.commit()print("转账成功")exceptExceptionase:# 任何一步出错，回滚conn.rollback()print(f"转账失败，已回滚：{e}")finally:conn.close()

2.2 关键点

• conn.commit()提交事务
• conn.rollback()回滚事务
• 出错必须回滚，否则已执行的 SQL 不会撤销
• 默认autocommit=False，所以需要手动提交

三、方式二：mysql-connector-python（官方驱动）

importmysql.connector conn=mysql.connector.connect(host='localhost',user='root',password='your_password',database='test_db')cursor=conn.cursor()try:cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 1")cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE user_id = 2")conn.commit()exceptmysql.connector.Erroraserr:conn.rollback()print(f"Error:{err}")finally:cursor.close()conn.close()

和pymysql逻辑一致，只是 API 略有不同。

四、方式三：SQLAlchemy ORM（推荐大型项目）

fromsqlalchemyimportcreate_engine,Column,Integer,Stringfromsqlalchemy.ormimportsessionmaker,declarative_base Base=declarative_base()classAccount(Base):__tablename__='accounts'id=Column(Integer,primary_key=True)name=Column(String(50))balance=Column(Integer)engine=create_engine('mysql+pymysql://root:password@localhost/test_db')Session=sessionmaker(bind=engine)session=Session()try:account1=session.query(Account).filter_by(id=1).with_for_update().first()account2=session.query(Account).filter_by(id=2).with_for_update().first()account1.balance-=100account2.balance+=100session.commit()print("转账成功")exceptExceptionase:session.rollback()print(f"转账失败：{e}")finally:session.close()

为什么用with_for_update()？

普通查询在并发下可能读到脏数据。with_for_update()会加行锁，确保这条记录在事务结束前不被其他事务修改，解决并发问题。

五、三种方式对比

选型建议：小项目用pymysql，追求稳定用官方驱动，项目大了直接上 SQLAlchemy。

六、常见坑 & 最佳实践

坑1：异常没捕获，事务没回滚

# ❌ 错误示范cursor.execute(sql1)cursor.execute(sql2)# 如果这里报错，sql1 已执行但没回滚conn.commit()

必须用 try/except 包裹，except 里调用rollback()。

坑2：连接池里的事务混乱

用连接池时，确保一个连接只处理一个事务，不要跨连接做事务操作。

坑3：忘了设置隔离级别

MySQL 默认隔离级别是REPEATABLE READ，但有些场景需要READ COMMITTED：

conn.begin()# 显式开启事务cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED")

最佳实践清单

• ✅ 事务里的 SQL 尽量少，减少锁持有时间
• ✅ 捕获所有异常，确保回滚
• ✅ 高并发场景加行锁（SELECT ... FOR UPDATE）
• ✅ 生产环境用连接池（如DBUtils、SQLAlchemy内置池）
• ✅ 不要在事务里做网络请求、文件 IO 等耗时操作

七、总结

事务不复杂，但用错了比不用更危险。记住三个动作：begin → commit / rollback → close，就能覆盖 90% 的场景。