Python使用MySQL事务的三种主流方式

在实际开发中，单条 SQL 往往不够用。转账、订单处理、库存扣减……这些场景要求多条 SQL **要么全成功，要么全失败**。这就是事务存在的意义。  这篇文章用 Python 实战讲解如何正确使用 MySQL 事务，覆盖 `pymysql`、`mysql-connector-python` 和 `SQLAlchemy` 三种主流方式。 ## 一、先搞懂事务的核心：ACID | 特性 | 含义 | 举例 | |------|------|------| | **A**tomicity（原子性） | 操作不可分割，全做或全不做 | 转账：扣款和入账必须同时成功 | | **C**onsistency（一致性） | 事务前后数据保持一致 | 余额不能凭空消失 | | **I**solation（隔离性） | 并发事务互不干扰 | 两人同时取钱，不能互相影响 | | **D**urability（持久性） | 提交后数据永久保存 | 服务器重启数据不丢失 | 一句话：**事务就是保证数据不出乱子的机制。** ## 二、方式一：pymysql（最常用） ### 2.1 基本用法 ```python import pymysql conn = pymysql.connect( host='localhost', user='root', password='your_password', database='test_db', charset='utf8mb4' ) try: with conn.cursor() as cursor: # 开启事务（默认就是手动提交模式） 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("转账成功") except Exception as e: # 任何一步出错，回滚 conn.rollback() print(f"转账失败，已回滚：{e}") finally: conn.close() ``` ### 2.2 关键点 - `conn.commit()` 提交事务 - `conn.rollback()` 回滚事务 - **出错必须回滚**，否则已执行的 SQL 不会撤销 - 默认 `autocommit=False`，所以需要手动提交 ## 三、方式二：mysql-connector-python（官方驱动） ```python import mysql.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() except mysql.connector.Error as err: conn.rollback() print(f"Error: {err}") finally: cursor.close() conn.close() ``` 和 `pymysql` 逻辑一致，只是 API 略有不同。 ## 四、方式三：SQLAlchemy ORM（推荐大型项目） ```python from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.orm import sessionmaker, declarative_base Base = declarative_base() class Account(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 -= 100 account2.balance += 100 session.commit() print("转账成功") except Exception as e: session.rollback() print(f"转账失败：{e}") finally: session.close() ``` ### 为什么用with_for_update()？ 普通查询在并发下可能读到脏数据。`with_for_update()` 会**加行锁**，确保这条记录在事务结束前不被其他事务修改，解决并发问题。 ## 五、三种方式对比 | 维度 | pymysql | mysql-connector | SQLAlchemy | |------|---------|----------------|------------| | 上手难度 | ⭐⭐ | ⭐⭐ | ⭐⭐⭐ | | 性能 | 快 | 快 | 稍慢（有 ORM 开销） | | 适用场景 | 轻量脚本、小项目 | 官方驱动、稳定需求 | 中大型项目 | | 并发控制 | 手动写 SQL | 手动写 SQL | `with_for_update()` 内置 | **选型建议**：小项目用 `pymysql`，追求稳定用官方驱动，项目大了直接上 SQLAlchemy。 ## 六、常见坑 & 最佳实践 ### 坑1：异常没捕获，事务没回滚 ```python # ❌ 错误示范 cursor.execute(sql1) cursor.execute(sql2) # 如果这里报错，sql1 已执行但没回滚 conn.commit() ``` **必须用 try/except 包裹，except 里调用 `rollback()`。** ### 坑2：连接池里的事务混乱 用连接池时，确保**一个连接只处理一个事务**，不要跨连接做事务操作。 ### 坑3：忘了设置隔离级别 MySQL 默认隔离级别是 `REPEATABLE READ`，但有些场景需要 `READ COMMITTED`： ```python conn.begin() # 显式开启事务 cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED") ``` ### 最佳实践清单 - ✅ 事务里的 SQL 尽量少，减少锁持有时间 - ✅ 捕获所有异常，确保回滚 - ✅ 高并发场景加行锁（`SELECT ... FOR UPDATE`） - ✅ 生产环境用连接池（如 `DBUtils`、`SQLAlchemy` 内置池） - ✅ 不要在事务里做网络请求、文件 IO 等耗时操作 ## 七、总结 | 你的场景 | 推荐方案 | |----------|----------| | 写个脚本批量处理数据 | `pymysql` + 手动 commit/rollback | | 官方项目，求稳定 | `mysql-connector-python` | | Web 项目、多人协作 | `SQLAlchemy` + `with_for_update()` | 事务不复杂，但**用错了比不用更危险**。记住三个动作：**begin → commit / rollback → close**，就能覆盖 90% 的场景。