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

2026-05-31 08:24:52 作者：detayun

本文详细讲解了如何在Python中使用MySQL事务,覆盖了pymysql、mysql-connector-python和SQLAlchemy三种主流方式,重点介绍了事务的ACID特性、常见用法及最佳实践,需要的朋友可以参考下

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

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

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

特性	含义	举例
Atomicity（原子性）	操作不可分割，全做或全不做	转账：扣款和入账必须同时成功
Consistency（一致性）	事务前后数据保持一致	余额不能凭空消失
Isolation（隔离性）	并发事务互不干扰	两人同时取钱，不能互相影响
Durability（持久性）	提交后数据永久保存	服务器重启数据不丢失

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

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

2.1 基本用法

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（官方驱动）

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（推荐大型项目）

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：异常没捕获，事务没回滚

# ❌ 错误示范
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 等耗时操作

七、总结

你的场景	推荐方案
写个脚本批量处理数据	`pymysql` + 手动 commit/rollback
官方项目，求稳定	`mysql-connector-python`
Web 项目、多人协作	`SQLAlchemy` + `with_for_update()`

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

以上就是Python使用MySQL事务的三种主流方式的详细内容，更多关于Python使用MySQL事务的资料请关注脚本之家其它相关文章！