PyMySQL连接数据库实战：连接参数、事务提交与批量插入性能避坑

脚本专家

PyMySQL是纯Python实现的MySQL驱动，安装不需要编译C扩展，在Windows、Linux、macOS上表现一致，API也高度兼容MySQLdb，支持Python 3全系，并能配合MySQL 8.x使用；原文还提到MariaDB在驱动层面同样兼容。下面围绕原文的用户表示例，按安装、连接参数、增删改查、事务、连接池和常见报错重组关键代码与排查路径。

一、环境安装与表结构准备
先安装并确认版本，避免多Python环境装错：

2. pip install pymysql
3. import pymysql
4. print(pymysql.__version__)

复制代码

在项目里可用requirements.txt锁版本，例如pymysql==1.1.0。写代码前先确认MySQL服务可用，并用最小权限账号连接。建库、建用户和演示表如下：

2. CREATE DATABASE demo_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
3. CREATE USER 'demo_user'@'localhost' IDENTIFIED BY 'your_password';
4. GRANT SELECT, INSERT, UPDATE, DELETE ON demo_db.* TO 'demo_user'@'localhost';
5. FLUSH PRIVILEGES;
6. USE demo_db;
7. CREATE TABLE users (
8.   id INT AUTO_INCREMENT PRIMARY KEY,
9.   username VARCHAR(64) NOT NULL,
10.   email VARCHAR(128),
11.   age INT,
12.   created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
13.   UNIQUE KEY uk_username (username)
14. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

复制代码

字符集统一用utf8mb4，因为MySQL的utf8实际是utf8mb3，无法完整支持emoji和生僻字。权限不要直接GRANT ALL。

二、连接参数与资源管理
最小连接示例如下，注意host使用127.0.0.1而不是localhost，部分系统会把localhost解析到IPv6的::1，而MySQL默认监听IPv4，容易出现Can't connect to MySQL server on 'localhost'。

2. import pymysql
3. conn = pymysql.connect(
4.   host='127.0.0.1',
5.   port=3306,
6.   user='demo_user',
7.   password='your_password',
8.   database='demo_db',
9.   charset='utf8mb4'
10. )
11. try:
12.   cursor = conn.cursor()
13.   cursor.execute('SELECT VERSION()')
14.   row = cursor.fetchone()
15.   print('MySQL版本：', row[0])
16. finally:
17.   cursor.close()
18.   conn.close()

复制代码

常用参数中，connect_timeout默认10秒，网络环境差时按需调整；max_allowed_packet默认67108864，插入大数据包时要调大；autocommit默认False，执行INSERT、UPDATE、DELETE后必须手动conn.commit()，否则看似执行成功但数据没有落库。该行为是事务保护机制，不是bug。PyMySQL连接和游标支持上下文管理，推荐用with自动释放：

2. with pymysql.connect(
3.   host='127.0.0.1',
4.   user='demo_user',
5.   password='your_password',
6.   database='demo_db',
7.   charset='utf8mb4'
8. ) as conn:
9.   with conn.cursor() as cursor:
10.     cursor.execute('SELECT id, username, email FROM users LIMIT 5')
11.     rows = cursor.fetchall()
12.     for row in rows:
13.       print(row)

复制代码

三、查询、分页与插入性能
游标提供fetchone()、fetchmany(size)、fetchall()。数据量小用fetchall没问题，几十万行全量加载可能撑爆内存，应使用fetchmany分块或服务端分页。分页查询示例：

2. page_size = 100
3. offset = 0
4. while True:
5.   cursor.execute(
6.     'SELECT id, username, email FROM users ORDER BY id LIMIT %s OFFSET %s',
7.     (page_size, offset)
8.   )
9.   rows = cursor.fetchall()
10.   if not rows:
11.     break
12.   for row in rows:
13.     print(row)
14.   offset += page_size

复制代码

插入单条可用cursor.lastrowid拿自增ID。批量插入优先executemany，原文本地实测1000条数据，单条循环约1.5秒，executemany约0.2秒，差距约7到10倍；十万级以上可考虑LOAD DATA LOCAL INFILE。示例：

2. sql = 'INSERT INTO users (username, email, age) VALUES (%s, %s, %s)'
3. data = []
4. for i in range(1000):
5.   data.append((f'user_{i}', f'user_{i}@example.com', 20 + i % 30))
6. cursor.executemany(sql, data)
7. conn.commit()
8. print('批量插入记录数：', cursor.rowcount)

复制代码

字段多时用DictCursor按字段名访问，便于JSON序列化和模板渲染；追求极致性能时默认元组游标略快，但万级以下差异可忽略。超大结果集用SSCursor或SSDictCursor，服务端保留游标、逐行拉取；同一连接上SSCursor是独占的，不能同时开第二个游标执行别的查询。

四、事务控制与自动提交选择
事务用于保证多条SQL要么全部成功、要么全部失败。PyMySQL中推荐try、commit、except、rollback结构：

2. try:
3.   conn.autocommit(False)
4.   cursor.execute('UPDATE users SET age = age + 1 WHERE username = %s', ('user_1',))
5.   cursor.execute('UPDATE users SET age = age - 1 WHERE username = %s', ('user_2',))
6.   conn.commit()
7.   print('事务提交成功')
8. except Exception as e:
9.   conn.rollback()
10.   print('事务回滚：', e)
11. finally:
12.   conn.close()

复制代码

autocommit=True适合纯查询、非关键数据写入、日志和统计上报；涉及多表写入、订单支付库存等强一致业务、批量导入需整体回滚时必须关闭自动提交。单条INSERT在InnoDB中也会被包成独立事务，关闭自动提交批量写入可减少redo log同步次数。

五、SQL注入与连接池
动态拼接SQL危险，用户输入会被当成SQL语法执行。正确做法是参数化占位符：

2. sql = 'SELECT * FROM users WHERE username = %s'
3. cursor.execute(sql, (username,))

复制代码

PyMySQL会把username当纯数据处理，不参与SQL语法解析。并发高时每次新建连接容易报Too many connections，可用DBUtils的PooledDB：

2. from dbutils.pooled_db import PooledDB
3. import pymysql
4. pool = PooledDB(
5.   creator=pymysql,
6.   maxconnections=6,
7.   mincached=2,
8.   maxcached=5,
9.   blocking=True,
10.   host='127.0.0.1',
11.   user='demo_user',
12.   password='your_password',
13.   database='demo_db',
14.   charset='utf8mb4'
15. )
16. conn = pool.connection()
17. cursor = conn.cursor()
18. cursor.execute('SELECT COUNT(*) FROM users')
19. print(cursor.fetchone())
20. conn.close()

复制代码

这里conn.close()是归还连接而非真正关闭；blocking=True表示池满时等待。原文压测中连接池比每次新建连接吞吐量约提升3倍。

六、常见报错与唯一键冲突
Can't connect通常查服务是否启动和host是否改127.0.0.1；Access denied查账号密码和GRANT权限；Unknown database查库名和database参数；Table doesn't exist查当前库和表名；Duplicate entry是唯一索引冲突；Incorrect string value多与字符集不支持emoji有关，表结构、连接参数和文件编码要统一utf8mb4；Lost connection和MySQL server has gone away要查max_allowed_packet、wait_timeout、空闲断开和重连保活。唯一键冲突可用一条SQL处理：

2. sql = '''
3. INSERT INTO users (username, email, age)
4. VALUES (%s, %s, %s)
5. ON DUPLICATE KEY UPDATE
6. email = VALUES(email),
7. agE = VALUES(age)
8. '''

复制代码

排查时建议按顺序确认：MySQL服务是否正常、账号权限是否足够、database和表名是否正确、字符集是否统一、autocommit是否忘记提交、连接数/超时/包大小是否触顶。上述代码与数据均来自原文示例，可按实际环境替换账号、密码和表名。