三年前一次线上事故，让我把 Python 连接 MySQL 这件事彻底重新学了一遍。业务一上线，某个订单模块就开始报pymysql.err.OperationalError: (1040, 'Too many connections')，MySQL 直接拒绝新连接，整个服务跟着雪崩。查到最后，原因特别朴素——代码里到处在pymysql.connect()，连接用完也不关，高并发一冲，数据库连接池直接被打满。
这个坑我相信很多团队都踩过。也正是从那次之后，我整理了一套 Python 操作 MySQL 的完整实践经验：驱动怎么选、连接怎么管理、事务怎么设计、批量数据怎么写、SQL 怎么防注入、典型报错怎么排查。这篇文章就是把这一堆内容系统化地讲一遍，适合已经会用 PyMySQL 写简单增删改查、但还没把这些“高级题目”吃透的同学。
1. 选型与连接参数：为什么我最终锁定了 PyMySQL
1.1 驱动对比：PyMySQL、MySQLdb 与 mysql-connector-python 的取舍
Python 连接 MySQL 的驱动有好几个，网上随便一搜就是各种安装教程和踩坑记录，但真正值得认真对比的就三个：MySQLdb（也就是mysqlclient）、mysql-connector-python、PyMySQL。
MySQLdb是 Python 2 时代的老牌驱动，C 扩展实现，性能确实好，但 Python 3 下的安装过程比较折腾，尤其在 Windows 上经常要编译或装额外的预编译包。对大多数业务场景来说，它带来的那点性能优势，根本比不上安装和环境带来的麻烦。
mysql-connector-python是 MySQL 官方提供的纯 Python 驱动，功能完整，文档也全，但它的 API 风格和社区里常见的写法有些差异，很多现成工具类库对它的支持也不如 PyMySQL 顺手。
PyMySQL是我最终的选择。它是纯 Python 实现，pip install pymysql一步到位，不需要编译，API 兼容MySQLdb，所以绝大多数的历史代码迁移成本极低。更关键的是，DBUtils、SQLAlchemy这些生态库对 PyMySQL 的适配都做得非常好，遇到问题搜解决方案也方便。
一句话总结：除非你有特别极端的性能需求或者团队历史包袱，否则新项目直接用 PyMySQL 就好。
1.2 connect() 里那些容易被忽略的关键参数
很多教程写连接只有五行：
import pymysql conn = pymysql.connect( host='127.0.0.1', port=3306, user='root', password='your_password', database='test_db' )这段代码能跑通，但离“生产可用”还差得远。我整理了一份完整的连接参数表，这些参数每一个都在线上环境里真实影响过系统的表现。
| 参数 | 默认值 | 建议值 | 什么时候会出问题 |
|---|---|---|---|
charset | 无 | 'utf8mb4' | 不设或设成utf8时，插入表情、生僻字报Incorrect string value |
cursorclass | 元组游标 | pymysql.cursors.DictCursor | 默认游标返回元组，字段一多就分不清哪个是哪个 |
autocommit | False | 根据业务定 | 忘了 commit 导致数据不落库，或忘了 rollback 导致连接状态脏掉 |
connect_timeout | 10 | 3~5 | 网络抖动时单次连接卡住数分钟 |
read_timeout | 无 | 10~30 | 慢 SQL 长时间占着连接不返回 |
write_timeout | 无 | 10~30 | 大批量写入时半路断连不报错 |
charset='utf8mb4'这句话我建议直接抄进每个项目的数据库连接代码里。MySQL 的utf8其实是utf8mb3，只支持最多 3 字节的字符，像 emoji 表情这种 4 字节字符直接就写不进去。买单号为了一张优惠券的 emoji 把订单创建接口搞挂，这种事故我见过不止一次。
cursorclass决定查询结果返回的形式。默认返回元组，row[0]、row[1]这样按下标取，字段多了以后可读性很差，而且 SQL 字段顺序一变，代码取值就错乱。用DictCursor，每一行是字典，row['order_id']这样取名，代码可维护性强一大截。
connect_timeout是很多人会忽略的。默认值是 10 秒，如果网络环境不好，一次连接失败可能要阻塞 10 秒才报错。在故障场景下，这个时间会被无限放大——因为每个请求都被卡住，积压的请求越来越多，服务就直接假死了。把它设成 3 到 5 秒，配合重试机制，故障恢复会快得多。
1.3 连接用完到底要不要手动 close
这个问题看起来基础，但真有不少人是靠程序退出时自动关闭连接的。用 PyMySQL 时，连接对象是一个真实的 TCP 连接加 MySQL 会话，Python 的垃圾回收不会立刻触发它关闭。如果你写的是长驻进程（比如 Web 服务），连接不手动关就是资源泄漏，超过max_connections后数据库直接拒绝新连接。
我个人的习惯是，所有的数据库连接操作都包在上下文管理器里：
from contextlib import contextmanager @contextmanager def get_connection(): conn = pymysql.connect( host='127.0.0.1', port=3306, user='root', password='your_password', database='test_db', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor, connect_timeout=5, read_timeout=15, write_timeout=15, ) try: yield conn finally: conn.close()finally里close()，无论业务代码有没有抛异常，连接都会被关掉。这个写法是底线，再往下就是用连接池了。
2. 连接池：一次线上事故让我明白连接复用有多重要
2.1 事故还原：连接数被打满的几个小时里发生了什么
先说那次事故。高峰期某个模块的接口开始报Too many connections，登录 MySQL 执行SHOW PROCESSLIST，看到几百个连接全部是Sleep状态。所谓Sleep，就是连接是活的、但什么也没干。显然，有大量连接被创建了以后就没有被释放。
查代码发现，某个功能为了省事，直接在每个请求里写了pymysql.connect()，用完了也不close()。为什么不会自动释放？Python 的对象引用计数 + 垃圾回收机制，在这种场景下根本不会及时回收连接对象，而且即使回收了，也需要时间。
这个问题的本质是：并发每上来一个请求，就多一个数据库连接，高峰期 1000 个请求同时进来，数据库瞬间就要创建上千个连接。MySQL 默认max_connections只有 151，直接被打爆。
2.2 一次数据库连接的隐性成本到底有多大
很多人以为数据库连接只是“网络连接一下，认证一下”，成本可不小。一次完整的 MySQL 连接建立过程包括：
- TCP 三次握手。
- MySQL 服务端进行身份认证、权限校验。
- 初始化会话级变量、加载用户权限缓存。
- 创建连接相关的内存结构。
整个过程在局域网环境下通常要 20~50 毫秒，跨机房或走公网要更久。单次连接看起来不算慢，但高并发场景下，连接建立的时间会被放大成巨大的开销，而且每个连接在 MySQL 服务端还需要额外的线程和内存资源。
拿一个实际的例子算一笔账：假设一个接口的数据库操作本身只要 10 毫秒，但连接建立要 30 毫秒。如果每次请求都新建连接，接口耗时直接从 10 毫秒变成 40 毫秒，慢了 4 倍。如果有连接池复用，这个 30 毫秒就完全省掉了。
2.3 用 DBUtils 搭一个稳妥的连接池
连接池的原理很简单：提前创建一批连接放在池子里，需要用的时候借一个，用完了还回去。Python 生态里最常用的连接池是DBUtils，它支持两种模式：PooledDB和PersistentDB。前者是真正的连接池，多个线程共用一批连接；后者是为每个线程维护一个专用连接。
我常用的配置是PooledDB：
from dbutils.pooled_db import PooledDB import pymysql pool = PooledDB( creator=pymysql, maxconnections=20, # 连接池允许的最大连接数 mincached=2, # 初始化时创建的最少空闲连接数 maxcached=10, # 最多缓存多少个空闲连接 blocking=True, # 连接数达到上限时，是否阻塞等待 ping=1, # 心跳检测，检查连接是否存活 host='127.0.0.1', port=3306, user='root', password='your_password', database='test_db', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor, )几个参数分别解释一下。
maxconnections是池子的“天花板”，设多少取决于数据库的max_connections和业务并发量。比如 MySQL 上限是 151，应用侧就不能设 200。一般我会给数据库上限预留 20% 到 30% 的余量，给运维和排查留空间。
blocking=True非常关键。如果连接被借完了，blocking=True会让请求排队等待而不是直接报错。虽然等待会拖慢响应，但至少不会让数据库因瞬间涌入的连接请求彻底崩溃。
ping=1表示每次借出连接时执行一个轻量的 ping 检测，连接如果已经因为wait_timeout被服务端断掉，就会重新建立连接。这个参数救过我很多次，尤其是业务上出现过空闲了几分钟再操作数据库报MySQL server has gone away的情况。
注意：老版本的DBUtils导入路径是from DBUtils.PooledDB import PooledDB，新版本（2.x 之后）改成了from dbutils.pooled_db import PooledDB。如果你安装后导入总是报错，先检查一下版本和路径。
2.4 连接池使用中必须养成的三个习惯
第一，用contextmanager把连接池封装成上下文管理器，业务代码只管“借”和“还”，不用管关闭。
from contextlib import contextmanager @contextmanager def get_conn_from_pool(): conn = pool.connection() try: yield conn conn.commit() # 没有异常时统一提交 except Exception: conn.rollback() # 有异常就回滚 raise finally: conn.close() # 归还连接，不是真正关闭注意，conn.close()在连接池里不是真的断开连接，而是把连接归还给池子。所以最终关闭连接的时机，是进程退出或者池子被销毁时，由PooledDB自己处理。
第二，不要在同一个业务操作里同时持有多个连接。有人写个简单的用户信息更新，先从连接池拿一个连接查用户，再从池子里拿另一个连接更新，两个连接都占着不放，稍微一并发，池子就枯竭了。正确做法是一个逻辑操作全程只用一个连接。
第三，事务逻辑要放在上下文管理器内部完成。如果你先with get_conn_from_pool()拿到了连接，然后退出这个上下文之后再去执行 SQL，连接已经归还，你以为自己在操作数据库，其实已经用的是另一条连接了，事务也就断开了。
3. 事务边界与隔离级别：数据一致性不是 commit 一下那么简单
3.1 autocommit 到底该谁说了算
autocommit这个参数，默认是False。也就是说，你执行完一条UPDATE之后，如果没有显式commit()，数据是不会真正落库的。这在写脚本时特别容易踩坑——脚本跑完了一查数据没变，原来是忘了提交。
反过来，如果把autocommit设为True，那每条 SQL 都是独立事务，自动提交。看起来省事了很多，但一旦一个业务操作需要多条 SQL 协同完成，autocommit=True就会变成灾难。
用一个特别常见的场景举例：用户下单扣库存。这个操作必须“查询库存是否足够 → 扣减库存 → 创建订单”，这三步是一个整体。如果每一步都自动提交，扣库存成功但创建订单失败，那用户的库存就白扣了。所以这个场景下autocommit必须设为False，手动控制事务边界。
3.2 从扣库存场景看事务的原子性
下面这段代码是一个标准的扣库存模板，特别注意事务的提交和回滚时机：
def deduct_stock(conn, product_id, quantity): try: with conn.cursor() as cursor: # 1. 查一下库存够不够 cursor.execute( "SELECT stock FROM products WHERE product_id = %s FOR UPDATE", (product_id,) ) row = cursor.fetchone() if not row or row['stock'] < quantity: raise ValueError("库存不足") # 2. 扣减库存 cursor.execute( "UPDATE products SET stock = stock - %s WHERE product_id = %s", (quantity, product_id) ) # 3. 创建订单 cursor.execute( "INSERT INTO orders (product_id, quantity) VALUES (%s, %s)", (product_id, quantity) ) conn.commit() except Exception: conn.rollback() raise这里有两个细节值得展开。
第一个，SELECT ... FOR UPDATE是行级锁，它会把这一行锁住，防止并发下两个请求同时读到同一个库存值。没有这行锁，两个请求同时查到库存为 1，都认为自己可以下单，结果就超卖了。
第二个，conn.rollback()放在except里清理事务，避免脏连接被归还到连接池后，下一个业务拿到一个未提交未回滚的会话。这个坑在连接池场景下尤其隐蔽——上一个请求回滚不及时，下一个请求复用同一个连接，事务状态是错的，数据就乱了。
3.3 隔离级别速查表与选型建议
SQL 标准定义了四种事务隔离级别，MySQL 默认是REPEATABLE READ（可重复读）。很多其他关系型数据库（比如 PostgreSQL）默认是READ COMMITTED。
| 隔离级别 | 脏读 | 不可重复读 | 幻读 | 特点 |
|---|---|---|---|---|
READ UNCOMMITTED | 可能 | 可能 | 可能 | 几乎不用，能读到未提交的数据 |
READ COMMITTED | 不可能 | 可能 | 可能 | 每次读到的都是已提交的数据 |
REPEATABLE READ | 不可能 | 不可能 | 可能 | MySQL 默认，同一事务内多次读结果一致 |
SERIALIZABLE | 不可能 | 不可能 | 不可能 | 用锁串行化，性能最差 |
简单解释一下这些概念：脏读是读到了别人还没提交的数据，别人回滚了你就读了个寂寞；不可重复读是同一事务里两次查询结果不一样，因为别的事务提交了修改；幻读是同一事务里两次范围查询返回的行数不一样，别的会话插入了新行。
MySQL 的REPEATABLE READ通过 MVCC（多版本并发控制）和间隙锁，在大部分场景下已经能规避幻读问题，所以大多数业务用默认级别是安全的。但如果你遇到大量死锁或者间隙锁导致的阻塞，可以考虑把隔离级别调到READ COMMITTED，减少锁的粒度。这在报表类、OLAP 类场景尤其常见。
修改会话级隔离级别：
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;不过我不建议轻易动全局配置，先评估下具体业务能不能接受“同一事务内可能读到不同的值”这个前提，再决定要不要降级。
3.4 一个可复用的 Python 事务模板
把事务和连接池整合起来，我会直接封装成一个装饰器或上下文管理器。这样业务代码里完全不需要关心事务的开始、提交和回滚。
from functools import wraps def transactional(func): @wraps(func) def wrapper(*args, **kwargs): with get_conn_from_pool() as conn: return func(conn, *args, **kwargs) return wrapper @transactional def create_order_with_stock_deduction(conn, product_id, quantity): # 这里面的所有 SQL 都在同一个事务里 deduct_stock(conn, product_id, quantity) # 其他业务逻辑...事务控制的核心原则是：事务一定要短。不要在事务里做网络请求、外部接口调用、耗时的本地计算，这些时间都会把行级锁或间隙锁攥在手里不放，导致其他事务排队。曾经见过一个同事在事务里调第三方支付接口，支付超时了 10 秒，数据库里这把锁就锁了 10 秒，直接连带影响一整批用户下单。
4. 批量写入与查询优化：性能差距从 10 倍起步
4.1 executemany 批量插入：别再一条一条 insert 了
大批量写入是 Python 操作 MySQL 时最容易出现性能问题的场景。最常见的错误写法是循环执行单条INSERT：
# 反面教材：1000 条数据循环插入 for item in item_list: cursor.execute( "INSERT INTO items (name, price) VALUES (%s, %s)", (item['name'], item['price']) ) conn.commit()每条execute都要走一次“SQL 解析 → 执行计划生成 → 写入 binlog → 刷盘”的流程，1000 条就是 1000 次完整流程。实际测试下来，即便在本地环境，逐条插入 1000 条数据通常也需要几百毫秒到数秒，网络越差越慢。
用executemany，一条调用就能批量提交：
data = [ ('商品A', 19.9), ('商品B', 29.9), ('商品C', 39.9), # ... 更多数据 ] cursor.executemany( "INSERT INTO items (name, price) VALUES (%s, %s)", data ) conn.commit()executemany会把多条数据合并成一条多值INSERT发送给 MySQL，减少网络往返和 SQL 解析次数。实测同样 1000 条数据，耗时可以降到逐条插入的十分之一，甚至更低。
还有一点要注意：不要一次性把几十万条数据全部塞给executemany。MySQL 对max_allowed_packet有上限，大包会被直接拒绝。稳妥的做法是分批执行，每批 5000 到 10000 条：
batch_size = 5000 for i in range(0, len(data), batch_size): batch = data[i:i + batch_size] cursor.executemany( "INSERT INTO items (name, price) VALUES (%s, %s)", batch ) conn.commit()4.2 流式游标与分批读取：百万行数据的正确姿势
默认游标会一次性把查询结果全部拉到客户端内存里。如果查询出来的是 500 万行，这些数据会全部堆积在内存里，很容易直接把进程打挂。
处理大数据集时用流式游标（SSCursor）。它只在数据库服务端保持游标，客户端一边遍历一边取数据，不会把所有结果一次性加载到内存。
import pymysql.cursors conn = pymysql.connect( host='127.0.0.1', user='root', password='your_password', database='test_db', charset='utf8mb4', cursorclass=pymysql.cursors.SSCursor, # 注意这里 ) with conn.cursor() as cursor: cursor.execute("SELECT id, name FROM large_table") while True: rows = cursor.fetchmany(1000) if not rows: break for row in rows: process(row)fetchmany(1000)每次只从网络读取 1000 行，处理完再取下一批。
有一个必须警惕的坑：SSCursor是流式的，它依赖连接持续从服务端拉取数据。如果你在SSCursor遍历过程中，又在同一个连接上执行了别的查询，会直接报错或者数据错乱。所以在用SSCursor时，这个连接就是“单任务”的，不能再干别的。
4.3 explain 一下，慢查询的原因立刻现形
慢查询优化，第一步永远是EXPLAIN。它不会真正执行 SQL，而是告诉你 MySQL 会怎么执行这条查询。
EXPLAIN SELECT * FROM orders WHERE user_id = 12345 ORDER BY created_at DESC LIMIT 10;输出结果里重点看几个字段：
| 字段 | 含义 | 需要警惕的情况 |
|---|---|---|
type | 访问类型 | ALL是全表扫描，性能最差 |
key | 实际使用的索引 | NULL说明没用上索引 |
rows | 预估扫描行数 | 越大越慢 |
Extra | 附加信息 | 出现Using filesort说明排序没走索引 |
举一个真实案例。某个订单查询接口，EXPLAIN结果显示type=ALL、rows=386512，一条查询两秒多。优化方案就是在user_id和created_at上加联合索引：
ALTER TABLE orders ADD INDEX idx_user_created (user_id, created_at);加索引之后再EXPLAIN，type变成ref，rows降到几百，查询时间从两千毫秒变成几十毫秒。这个提升非常直接，也是热搜词里大量出现“mysql 排序”“mysql join 含义”的根本原因——ORDER BY排序字段如果没有索引，MySQL 就要把查出来的数据全部放进临时文件里做filesort；JOIN的关联字段如果没索引，就要对驱动表全表扫描，这些都是表面看不到的隐性成本。
优化排序和 join 的原则很朴素：WHERE条件字段、ORDER BY字段、JOIN关联字段，尽量都建合适的索引。但别盲目加索引，索引也是一把双刃剑——每个索引都会拖慢写入，因为每次INSERT、UPDATE、DELETE都要同步维护索引。业务上不常查的字段，给再多索引都是负担。
5. 参数化查询与存储过程：安全和高阶玩法
5.1 用户输入直接拼接 SQL 的代价
SQL 注入攻击，本质上是用户输入被当成 SQL 代码执行了。最典型的写法长这样：
# 反面教材 username = request.form['username'] password = request.form['password'] sql = f"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'" cursor.execute(sql)如果用户在用户名里输入' OR '1'='1，拼接出来的 SQL 就变成了：
SELECT * FROM users WHERE username = '' OR '1'='1' AND password = 'xxx'OR '1'='1'恒为真，整条查询直接返回所有用户记录。这只是最简单的一种注入姿势，更狠的还能通过分号拼接DELETE、DROP之类的语句。
参数化查询可以从根上解决这个问题：
sql = "SELECT * FROM users WHERE username = %s AND password = %s" cursor.execute(sql, (username, password))参数化查询的原理是：SQL 语句结构和参数值分两条通道发给 MySQL，参数值不会被当成 SQL 代码解析，而是作为纯文本值处理。这就是为什么%s占位符是 Python 数据库操作中的铁律——凡是涉及用户输入的地方，一律参数化，没有任何例外。
搜“mysql update 语法”“mysql 数据库命令大全”的同学，网上很多资料都会提到字符串拼接的写法，那些历史教程里的写法在今天的生产环境里是要吃大亏的。不管是在 PyMySQL、SQLAlchemy 还是其他 ORM 里，都该用参数绑定。
5.2 callproc() 调用存储过程的姿势与坑
热搜词里有“mysql 声明存储过程”“mysql 存储过程”，说明很多人正在接触存储过程。Python 调用存储过程，PyMySQL 提供了callproc方法。
假设一个已经存在的存储过程，传入用户 ID 返回用户订单总数：
CREATE PROCEDURE get_order_count(IN uid INT, OUT total INT) BEGIN SELECT COUNT(*) INTO total FROM orders WHERE user_id = uid; ENDPython 端调用：
cursor.callproc('get_order_count', (12345, 0)) # 存储过程执行完，返回的是普通结果集 # OUT 参数的值，可以通过 SELECT @_变量名 来取 cursor.execute("SELECT @_get_order_count_1") result = cursor.fetchone() print(result) # {'@_get_order_count_1': 12}注意@_get_order_count_1这种名字的规则：@_存储过程名_参数序号。序号从 0 开始，所以第一个参数uid对应@_get_order_count_0，第二个参数total对应@_get_order_count_1。
还有一个容易踩的坑：如果存储过程内部有多个查询语句（比如临时表、中间结果集），callproc之后需要把多个结果集都 fetch 完，否则连接状态会残留，影响下一次操作。这是存储过程调用里最隐蔽的问题之一。
至于到底要不要用存储过程，我的个人看法是：对于简单的数据操作，业务逻辑放应用层更好，毕竟代码更好维护、更容易测试；但如果是复杂报表统计、定时任务里的大量数据加工，或者是要强制统一数据口径的场景，存储过程有它不可替代的优势——比如你在多个服务里都要算同一个“用户月度消费金额”指标，直接用同一个存储过程，比每个服务各写一套逻辑要可靠得多。
5.3 类型映射细节：Decimal、datetime 与 JSON
Python 和 MySQL 的数据类型不是一一对应的，这里面有几个常见的“暗坑”。
| MySQL 类型 | Python 类型 | 坑点 |
|---|---|---|
DECIMAL | decimal.Decimal | 不能直接转成float，会有精度丢失 |
DATETIME/TIMESTAMP | datetime.datetime | 时区问题，最好统一用 UTC 或统一 Asia/Shanghai |
TINYINT(1) | int（0/1） | 注意区分布尔值和数字 |
JSON | str | 需要自己json.loads()解析 |
BIGINT | int | Python3 里无溢出问题，不用特殊处理 |
涉及钱的数据，最痛的就是DECIMAL。MySQL 里的DECIMAL(10,2)在 Python 里返回Decimal('19.90')，这是精确的十进制数。如果你手贱执行float(Decimal('19.90'))，它永远不会精确等于 19.9，后续再乘数量、算折扣，结果就会带上诡异的浮点误差，比如 19.899999。金额运算要全程用Decimal，到最后展示再转字符串。
JSON 字段是另一个坑。MySQL 5.7 之后提供原生 JSON 类型，但 PyMySQL 拿到的是字符串，不是 dict。所以从数据库读 JSON 字段后，需要：
import json row = cursor.fetchone() config = json.loads(row['config_data'])6. 高频报错的排查思路：从 error 2002 到 gone away
6.1 error 2002：TCP 连不上和 socket 连不上的区别
error 2002 (HY000): Can't connect to local MySQL server through socket '/tmp/mysql.sock'是搜索量很高的经典报错。用 PyMySQL 连接时，如果你把host写成localhost，驱动会尝试通过 Unix socket 文件连数据库，而不是走 TCP/IP 协议。
排查链路按这四步走：
第一步，确认 MySQL 服务真的在运行。Linux 上执行：
systemctl status mysql # 或者 systemctl status mysqld如果服务是停了，启动它就行：systemctl start mysql。
第二步，确认 socket 文件到底在哪。MySQL 的 socket 路径不一定都是/tmp/mysql.sock，很多发行版配置成/var/run/mysqld/mysqld.sock。用下面的 SQL 查：
SHOW VARIABLES LIKE 'socket';第三步，在 Python 连接参数里显式加上 socket 路径，或者干脆改用127.0.0.1走 TCP：
conn = pymysql.connect( host='127.0.0.1', # 用 IP 而不是 localhost port=3306, user='root', password='your_password', database='test_db', unix_socket='/var/run/mysqld/mysqld.sock', # 或这里指定 socket charset='utf8mb4', )第四步，检查 socket 文件的权限。如果 MySQL 服务正常、路径也对了，但权限不足，也是同样的报错。用ls -l /var/run/mysqld/mysqld.sock看属主和权限，确保 Python 进程的用户有权限访问。
经验之谈：开发环境图省事可以直接用
127.0.0.1跳过 socket 问题，但生产环境要明确到底用 TCP 还是 socket，两种方式在认证和权限模型上有差异，混用容易出怪问题。
6.2 too many connections：连接数被占满的快速处置
pymysql.err.OperationalError: (1040, 'Too many connections')的排查思路，在本文开头的事故里已经展示过，这里把标准处置步骤整理清楚。
第一步，确认当前连接数：
SHOW STATUS LIKE 'Threads_connected'; SHOW VARIABLES LIKE 'max_connections';第二步，查看哪些连接占着不放：
SHOW PROCESSLIST;重点看Command列，如果大量是Sleep，说明连接借出去后没有被归还或关闭。找到具体来源后，可以考虑杀掉一部分空闲连接，先让服务恢复：
-- 把 sleep 超过 100 秒的连接查出来，再逐个 KILL SELECT id, user, host, db, command, time FROM information_schema.processlist WHERE command = 'Sleep' AND time > 100;第三步，从根源上解决。连接数被打满几乎都是应用层连接管理出了问题：没关连接、没用连接池、连接池配置过小。临时调大max_connections没有意义，它只是把问题往后拖延了几分钟，真正要改的是代码。
6.3 MySQL server has gone away 的前因后果
这个报错的意思是：MySQL 服务端把这个连接断开了，但客户端还在用它。常见原因有三个。
一是连接空闲时间超过了wait_timeout。MySQL 默认wait_timeout是 28800 秒（8 小时），如果连接在池子里空闲太久，服务端会主动断开。解法就是连接池的ping参数，像 2.3 节那样设置ping=1，每次借出连接前先做存活检测。
二是max_allowed_packet太小。当你一次性写入的数据包超过这个值（默认是 67108864 字节，即 64MB），MySQL 会直接断开连接。排查方法是看 MySQL 错误日志，里面有packet too large之类的记录。
三是网络层面的问题。比如数据库在别的机房，网络设备清掉空闲连接、防火墙把长时间没有流量的 TCP 连接回收了，都会导致连接在空闲一段时间后失效。解法是连接池定期做心跳，或者写一个重试机制。
我常用一个简单的自动重试装饰器，把“断线自动重连”做成通用逻辑：
import time import pymysql RETRYABLE_ERRORS = (pymysql.err.OperationalError,) def retry_on_db_error(max_retries=3, delay=0.5): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_retries): try: return func(*args, **kwargs) except RETRYABLE_ERRORS as e: if attempt == max_retries - 1: raise time.sleep(delay * (attempt + 1)) return None return wrapper return decorator这个重试装饰器只适用于“连接断掉、事务未开始或已回滚”的情况；如果错误发生在事务进行中，直接重试会造成重复更新之类的逻辑错误。所以重试之前要确保事务状态是干净的。
最后再分享一个我自己长期在用的习惯：线上环境的数据库操作，一定要配合慢查询日志看性能表现。PyMySQL 端把cursor.execute()的执行时间打到日志里，发现执行超过 500 毫秒的 SQL 就拿出来EXPLAIN分析。这个习惯帮我提前抓出过不少索引失效和慢排序的问题。数据库方面的“高级操作”，说到底就是连接管理、事务边界、索引设计这几件事的排列组合，把每一件都盯住了，线上基本不会出大问题。
