Python操作MySQL数据库时常见的三个陷阱及解决方法
作者：detayun
坑一：pymysql —— 异常吞掉，钱扣了没到账
案发现场
import pymysql
conn = pymysql.connect(host='localhost', user='root', password='pwd', database='test')
cursor = conn.cursor()
# 转账：扣款
cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
# 假设这里程序崩溃 / 网络超时 / 抛出异常...
# 加款永远没执行
cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
conn.commit() # 第一条已经生效，无法回滚
结果：用户A被扣了100块，用户B没收到钱。数据不一致，且无法挽回。
根因
pymysql 默认 autocommit=False，但每条 execute 本身不会自动提交——问题出在异常发生后，第一条 SQL 已经被 MySQL 隐式提交（因为 pymysql 在某些异常场景下会触发自动提交），或者更常见的情况是：开发者以为异常会自动回滚，但实际上没有显式调用 rollback()。
解决方法
try:
cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
conn.commit()
except Exception as e:
conn.rollback() # ← 必须显式回滚
raise # 或记录日志后重新抛出
finally:
cursor.close()
conn.close()
关键原则：try 里写业务 SQL，except 里必须有 rollback()，finally 里关闭连接。三件套缺一不可。
坑二：mysql-connector-python —— 连接池泄漏，服务跑着跑着就挂了
案发现场
import mysql.connector
from mysql.connector import pooling
dbconfig = {
"host": "localhost", "user": "root", "password": "pwd", "database": "test"
}
connection_pool = pooling.MySQLConnectionPool(pool_name="mypool", pool_size=5, **dbconfig)
def get_user(user_id):
conn = connection_pool.get_connection() # 从池中取连接
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
result = cursor.fetchall()
# 忘记关闭连接，直接返回
return result
结果：前5次请求正常，第6次开始报错 Too many connections。因为连接取走后没归还，池子里的5个连接被永久占用。
根因
mysql-connector-python 的连接池行为是连接常驻——取走后不会自动回收。与 SQLAlchemy 的 QueuePool（空闲超时后销毁连接）不同，这里的连接一旦泄漏，池子容量就永久减少，直到耗尽。
解决方法
def get_user(user_id):
conn = connection_pool.get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
return cursor.fetchall()
finally:
cursor.close()
conn.close() # ← 必须归还连接到池中
或者用上下文管理器：
from contextlib import closing
def get_user(user_id):
with closing(connection_pool.get_connection()) as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
return cursor.fetchall()
# 退出 with 自动关闭，连接归还池中
监控指标：定期执行 SHOW PROCESSLIST，观察 Sleep 状态的连接数是否持续增长。
坑三：SQLAlchemy —— 批量更新时，事件监听器"沉默"了
案发现场
from sqlalchemy import event, update
from sqlalchemy.orm import Session
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(50))
updated_at = Column(DateTime)
# 定义事件：每次更新自动维护 updated_at
@event.listens_for(User, 'before_update')
def set_updated_at(mapper, connection, target):
target.updated_at = datetime.now()
# 业务代码：批量更新用户状态
session = Session()
stmt = update(User).where(User.status == 'pending').values(status='processed')
session.execute(stmt) # ← 监听器根本没触发
session.commit()
结果：1000条记录的 status 改了，但 updated_at 全是 NULL。因为 session.execute(stmt) 走的是 Core 层，绕过了 ORM 的对象生命周期，事件监听器根本不知道发生了更新。
根因
SQLAlchemy 的事件系统绑定的是 ORM 对象的状态变化。session.execute(update(...)) 是直接执行 SQL，不经过 ORM 对象加载，所以 before_update / after_update 全部静默。
解决方法（三选一）
| 方案 | 适用场景 | 代码 |
|---|---|---|
| 回归 ORM 遍历更新 | 数据量 < 1万 | for u in session.query(User).filter_by(status='pending'): u.status='processed' |
| SQL 里直接写 | 追求性能 | values(status='processed', updated_at=func.now()) |
| 数据库触发器 | 强一致性要求 | 交给 MySQL 层，ON UPDATE CURRENT_TIMESTAMP |
推荐大多数场景用方案二，性能和一致性兼顾：
from sqlalchemy import func
stmt = (
update(User)
.where(User.status == 'pending')
.values(status='processed', updated_at=func.now())
)
session.execute(stmt)
session.commit()
三个坑的本质对比
| 坑一 pymysql | 坑二 mysql-connector | 坑三 SQLAlchemy | |
|---|---|---|---|
| 表面问题 | 异常没回滚 | 连接耗尽 | 事件没触发 |
| 真实原因 | 异常处理不完整 | 连接池机制不同 | Core vs ORM 路径差异 |
| 一句话解法 | except 里必写 rollback() | finally 里必关连接 | 批量操作别用 Core 绕过 ORM |
这三个坑覆盖了 异常流、资源管理、抽象层边界 三个维度，踩过一次就不会再踩第二次。
以上就是Python操作MySQL数据库时常见的三个陷阱及解决方法的详细内容，更多关于Python操作MySQL数据库常见陷阱的资料请关注脚本之家其它相关文章！