PyMySQL入门到实战：连接管理、事务处理与性能优化指南

脚本专家

在Python Web开发和数据处理中，与MySQL数据库的交互几乎成为标配。PyMySQL作为一款纯Python实现的MySQL驱动，凭借其简单安装、良好兼容性和原生Python 3支持，已成为开发者操作MySQL的首选工具。本文将从实际使用中的痛点出发，深入讲解PyMySQL的连接管理、SQL注入防范、事务处理、生产环境最佳实践，并附上完整的代码示例。

一、为什么选择PyMySQL
常见的Python连接MySQL方案包括MySQLdb（仅支持Python 2，已淘汰）、mysql-connector-python（官方出品但因纯Python性能一般）、PyMySQL（纯Python实现，兼容MySQLdb接口）以及上层ORM框架如SQLAlchemy。PyMySQL无需编译C扩展，安装方便，跨平台，接口与MySQLdb高度一致，迁移成本低，且社区活跃持续更新。如果你的项目需要直接与MySQL交互而不引入复杂的ORM，PyMySQL是最实用的选择。

二、核心问题与PyMySQL的解决方案

• 连接管理：手动创建连接后容易忘记关闭，导致连接泄漏。PyMySQL支持使用 `with` 上下文管理器自动管理连接的打开与关闭，从根本上杜绝泄漏。
  
• SQL注入风险：直接使用字符串拼接SQL语句极易被注入攻击。PyMySQL提供参数化查询，通过 `%s` 占位符传递参数，库内部自动转义特殊字符。
  
• 事务处理：转账等场景需保证数据一致性，必须正确使用事务的提交和回滚。PyMySQL通过 `conn.commit()` 和 `conn.rollback()` 让开发者掌控事务边界。
  

三、安装与版本要求
运行 `pip install PyMySQL` 即可安装，如需指定版本：`pip install PyMySQL==1.1.0`，建议使用国内镜像加速。环境要求 Python 3.7+ 以及 MySQL 5.7+ 或 MariaDB 10.2+。安装后通过 `import pymysql; print(pymysql.__version__)` 验证。

四、连接与游标：基础概念
连接（Connection）类似于程序与数据库之间的通信通道，需要通过主机、端口、用户名、密码、数据库名、字符集等参数建立。游标（Cursor）是执行SQL语句和获取结果的对象。实际操作中推荐将连接和游标都放入 `with` 语句块中，确保异常时也能正确释放资源。

2. import pymysql
4. with pymysql.connect(
5.     host='localhost',
6.     port=3306,
7.     user='root',
8.     password='your_password',
9.     database='test_db',
10.     charset='utf8mb4'
11. ) as connection:
12.     with connection.cursor() as cursor:
13.         cursor.execute("SELECT VERSION()")
14.         version = cursor.fetchone()
15.         print(f"MySQL版本: {version[0]}")
16. # 离开 with 块后连接自动关闭

复制代码

五、基础数据操作

• 查询数据：使用 `cursor.execute()` 执行SELECT语句，通过 `fetchone()` 获取单条记录，`fetchall()` 获取全部记录。建议使用 `pymysql.cursors.DictCursor` 游标类型，返回字典格式（如 `user['name']`）而非元组，代码可读性更高。
  
• 插入数据：使用参数化插入 `INSERT INTO users (name, email) VALUES (%s, %s)`，参数以元组形式传入 `execute()`。批量插入使用 `executemany(sql, data_list)`，其中 `data_list` 是多个参数元组的列表。注意插入后必须调用 `conn.commit()` 提交事务。
  
• 更新和删除：同样使用参数化SQL，在 `execute()` 后调用 `commit()`。通过 `cursor.rowcount` 获取影响行数。
  

2. from pymysql.cursors import DictCursor
4. with pymysql.connect(..., cursorclass=DictCursor) as conn:
5.     with conn.cursor() as cursor:
6.         cursor.execute("SELECT id, name FROM users")
7.         users = cursor.fetchall()
8.         for user in users:
9.             print(user['id'], user['name'])

复制代码

六、事务管理：保证数据一致性
以银行转账为例，必须确保扣款和加款同时成功或同时失败。典型做法是：在try块中依次执行多个更新语句，全部成功后调用 `conn.commit()`；一旦发生异常，在except块中调用 `conn.rollback()` 撤销所有变更。PyMySQL的 `MySQLError` 可捕获数据库相关错误。

2. from pymysql import MySQLError
4. def transfer(from_id, to_id, amount):
5.     with pymysql.connect(...) as conn:
6.         try:
7.             with conn.cursor() as cursor:
8.                 # 检查余额、扣款、加款
9.                 cursor.execute("SELECT balance FROM accounts WHERE id=%s", (from_id,))
10.                 if cursor.fetchone()[0] < amount:
11.                     return False
12.                 cursor.execute("UPDATE accounts SET balance=balance-%s WHERE id=%s", (amount, from_id))
13.                 cursor.execute("UPDATE accounts SET balance=balance+%s WHERE id=%s", (amount, to_id))
14.                 conn.commit()
15.                 return True
16.         except MySQLError as e:
17.             conn.rollback()
18.             print(f"回滚: {e}")
19.             return False

复制代码

七、参数化查询与防SQL注入
永远不要使用 `f"SELECT * FROM users WHERE name = '{name}'"` 这种字符串拼接方式，否则如果 `name` 包含 `'; DROP TABLE users; --` 会造成灾难。正确的做法是用占位符 `%s` 并传入参数元组，PyMySQL会自动转义特殊字符。

2. sql = "SELECT * FROM users WHERE name = %s"
3. cursor.execute(sql, (user_input_name,))

复制代码

八、游标类型选择
PyMySQL提供多种游标类型：

• `Cursor`：默认，返回元组。
  
• `DictCursor`：返回字典，推荐用于可读性。
  
• `SSCursor`：服务端游标，结果集不一次性加载到客户端，适合处理百万级数据，需逐行迭代。
  
• `SSDictCursor`：服务端字典游标，兼顾大数据量和字典访问。
  

九、生产环境最佳实践

• 连接池：使用 `DBUtils` 的 `PooledDB` 创建连接池，避免频繁创建/销毁连接。常用参数：`maxconnections`（最大连接数，根据并发设置10-50）、`mincached`（初始化最小空闲连接2-5）、`blocking=True`（池满时阻塞等待）。
  
• 超时配置：在 `pymysql.connect()` 中设置 `connect_timeout=10`（连接超时）、`read_timeout=30`、`write_timeout=30`，防止长时间阻塞。`autocommit` 根据业务是否需要手动事务设为 `False`。
  
• 异常处理：捕获 `OperationalError`（连接问题）、`ProgrammingError`（SQL语法）、`IntegrityError`（主键冲突等）、`DataError`（数据越界）等，并给出相应处理（如重连、记录日志）。
  
• 封装工具类：将常用操作如 `query()`, `execute()`, `execute_many()`, `execute_with_transaction()` 封装成类，简化业务代码。可基于 `contextmanager` 实现连接上下文管理器。
  

2. from dbutils.pooled_db import PooledDB
3. pool = PooledDB(creator=pymysql, maxconnections=20, mincached=5, maxcached=10,
4.                 blocking=True, host='localhost', user='root', password='...', database='test')
5. def get_user(user_id):
6.     conn = pool.connection()
7.     try:
8.         with conn.cursor(DictCursor) as cursor:
9.             cursor.execute("SELECT * FROM users WHERE id=%s", (user_id,))
10.             return cursor.fetchone()
11.     finally:
12.         conn.close()  # 归还到池中

复制代码

十、总结
掌握PyMySQL的核心在于理解连接与游标的生命周期管理、坚持参数化查询、正确使用事务，以及在生产环境中借助连接池和超时配置提升稳定性。通过将常用操作封装为工具类，可以极大简化数据库交互代码，减少重复劳动。无论你是初学者还是正在构建大型Python服务，PyMySQL都是一个可靠且高效的底层数据库驱动。

热心网友5

感谢楼主的详细教程！内容非常实用，特别是事务管理和参数化查询的部分，正好解决了我最近写转账接口时遇到的疑惑。想问一下，实际生产环境中你一般会设置连接池吗？或者使用 with 语句自动管理连接在高并发下够用吗？