Python连接MySQL实战：PyMySQL操作、事务与连接池全解析

1. 先搞明白：为什么是PyMySQL

在日常的后端开发、数据处理脚本、自动化运维任务里，和数据库打交道是躲不开的环节。Python这边连接MySQL的库有好几个，老牌的MySQLdb、SQLAlchemy这种ORM，还有今天要细聊的PyMySQL。我自己的主力选择就是PyMySQL，原因很直接：它纯Python实现，安装不需要编译C扩展，在Windows、Linux、macOS上表现一致，出问题也好排查。

更关键的一点，PyMySQL的API设计高度兼容MySQLdb，如果你是从旧项目迁移过来，代码改动量非常小。它支持Python 3全系，配合最新的MySQL 8.x版本也毫无压力。这个库解决的问题很简单：让Python程序能够稳定、高效地和MySQL数据库对话，执行查询、插入、更新、删除，管理事务，处理异常。

这篇内容适合谁看？刚入门Python想连数据库的新手，被各种连接报错折磨的排查者，需要在脚本里批量处理数据的开发者，以及想搞清楚事务和连接池原理的进阶玩家。我会把从安装到实战、从连接到优化、从常用操作到避坑经验都过一遍，内容偏实操，直接能抄。

提示：如果你用的是MariaDB，PyMySQL也同样支持，它在驱动层面的兼容性是没问题的。

2. 准备工作：环境安装与依赖梳理

2.1 快速安装PyMySQL

安装非常简单，pip一把梭：

pip install pymysql

如果你在多个Python环境间切换，或者要用虚拟环境，先激活再装，别装错环境。装了多个Python版本的老手经常遇到“明明装了我却导入不了”的尴尬，就是环境串了。

验证是否装好，进Python交互环境试一下：

import pymysql print(pymysql.__version__)

能输出版本号就说明安装成功。我这边习惯在项目里用requirements.txt锁版本，避免过阵子库更新导致行为变化：

pymysql==1.1.0

2.2 确认MySQL服务可用

在写Python代码之前，先确认MySQL本身是活着的。这一步很多人跳过，然后代码报错半天找不到原因，最后发现是MySQL服务没启动。

命令行试连接：

mysql -u root -p

能进去，说明服务正常。接下来在MySQL里创建一个用于测试的数据库和账号，我建议别拿root账号直接跑业务代码，最小权限原则什么时候都不过时：

CREATE DATABASE demo_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE USER 'demo_user'@'localhost' IDENTIFIED BY 'your_password'; GRANT SELECT, INSERT, UPDATE, DELETE ON demo_db.* TO 'demo_user'@'localhost'; FLUSH PRIVILEGES;

这里有几个关键点说明一下：

• 字符集用utf8mb4而不是utf8，因为utf8mb4能完整支持emoji和生僻字，MySQL的utf8实际是utf8mb3，很多情况会报字符集转换错误。
• 权限按需给，不要一上来GRANT ALL，生产环境出过太多因为权限过宽导致的线上事故。
• FLUSH PRIVILEGES确保权限立即生效。

2.3 设计一张用于演示的数据表

后续所有代码示例都围绕用户表展开，先建表：

USE demo_db; CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(64) NOT NULL, email VARCHAR(128), age INT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_username (username) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

这张表包含了自增主键、唯一索引、默认值时间戳，足够覆盖大部分入门场景。建完表之后，我们就能把注意力完全放到Python代码上了。

3. 核心连接逻辑与关键参数

3.1 最朴素的连接方式

PyMySQL的connect方法是最常用的入口。来一段最小可用示例：

import pymysql conn = pymysql.connect( host='127.0.0.1', port=3306, user='demo_user', password='your_password', database='demo_db', charset='utf8mb4' ) try: # 创建游标 cursor = conn.cursor() # 执行一条查询，确认连接是通的 cursor.execute("SELECT VERSION()") row = cursor.fetchone() print("MySQL版本：", row[0]) finally: cursor.close() conn.close()

注意几个细节：

• host用的127.0.0.1而不是localhost，在部分系统上，Python解析localhost会走IPv6（::1），MySQL默认监听IPv4，就容易踩Can't connect to MySQL server on 'localhost'的坑。用127.0.0.1能避免绝大多数这类问题。
• 端口3306是MySQL默认端口，如果你改了配置，这里要跟着改。
• 游标和连接用完必须关闭，Python的finally块就是干这个的。不关的话，连接会一直占着，数据库连接数飙升，最终服务拒绝新连接。

3.2 连接参数背后的逻辑

既然要深入，把常用连接参数讲透，以后用什么都受益：

有一个容易被忽略的点：PyMySQL默认autocommit=False，也就是说你执行INSERT、UPDATE、DELETE之后必须手动conn.commit()，否则数据不会真正落库。新手最容易在这里迷糊，明明执行没报错，一看表里数据没变，就是因为没提交事务。这个行为是保护机制，不是bug，配合事务使用才是正确的姿势。

3.3 使用with语法简化连接管理

Python的with语法能帮我们自动关闭资源，让代码更整洁。PyMySQL连接对象支持上下文管理协议：

from contextlib import closing import pymysql with pymysql.connect( host='127.0.0.1', user='demo_user', password='your_password', database='demo_db', charset='utf8mb4' ) as conn: with conn.cursor() as cursor: cursor.execute("SELECT id, username, email FROM users LIMIT 5") rows = cursor.fetchall() for row in rows: print(row)

用with conn块包住连接，退出块时连接会自动关闭，不用担心忘了释放。游标同样支持上下文管理，这是推荐写法，代码简洁且安全。

4. 增删改查全流程实操

4.1 查询数据：fetch系列方法详解

查询是最常用的操作。PyMySQL游标提供了三个fetch方法：

• fetchone()：取一行
• fetchmany(size)：取指定行数
• fetchall()：取剩余全部

它们的使用区别要搞清楚，特别是内存消耗。数据量小用fetchall()无所谓，但查出来几十万行，全部塞进内存，Python进程可能会直接内存溢出。正确的做法是用fetchmany(size)分块取，或者直接让MySQL分页。

来一个分页查询的例子：

import pymysql conn = pymysql.connect( host='127.0.0.1', user='demo_user', password='your_password', database='demo_db', charset='utf8mb4' ) cursor = conn.cursor() # 服务端分页，避免全量加载到内存 page_size = 100 offset = 0 while True: cursor.execute( "SELECT id, username, email FROM users ORDER BY id LIMIT %s OFFSET %s", (page_size, offset) ) rows = cursor.fetchall() if not rows: break for row in rows: print(row) offset += page_size cursor.close() conn.close()

这里使用参数化查询%s占位符，千万别用字符串拼接SQL，SQL注入事故就是这么来的。参数化查询不光是安全问题，它对SQL缓存命中也有好处。

4.2 插入数据：单条与批量对比

插入单条记录：

import pymysql conn = pymysql.connect( host='127.0.0.1', user='demo_user', password='your_password', database='demo_db', charset='utf8mb4', autocommit=True ) cursor = conn.cursor() sql = "INSERT INTO users (username, email, age) VALUES (%s, %s, %s)" cursor.execute(sql, ('alice', 'alice@example.com', 28)) print("插入ID：", cursor.lastrowid) conn.close()

cursor.lastrowid拿自增ID。如果设置了autocommit=False，需要手动执行conn.commit()。

批量插入的效率对比可能是大家最关心的。直接循环单条插入1000条有多慢？每条一次网络往返，1000次往返，延迟全耗在网络上。更好的方式是executemany：

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

我在本地测过，1000条数据，单条循环插入耗时大概1.5秒上下，executemany只用了0.2秒左右，差距基本在7-10倍。数据量越大，executemany优势越明显。

注意：executemany底层也是循环执行，只是省去了Python到MySQL的重复解析开销。如果插入十万级以上数据，更推荐用LOAD DATA LOCAL INFILE，那个是真正的高速批量导入工具，效率还能再翻几倍。

4.3 列表转字典：DictCursor的妙用

默认游标返回元组，字段多了以后靠下标访问很容易错位。比如row[2]到底是email还是age，没人记得住。PyMySQL自带DictCursor，直接用字段名访问：

import pymysql from pymysql.cursors import DictCursor conn = pymysql.connect( host='127.0.0.1', user='demo_user', password='your_password', database='demo_db', charset='utf8mb4', cursorclass=DictCursor ) cursor = conn.cursor() cursor.execute("SELECT id, username, email FROM users WHERE id = %s", (1,)) row = cursor.fetchone() print(row['username']) # 直接用字段名，可读性高很多 conn.close()

这是我在项目里常用的设置，返回字典之后，JSON序列化、模板渲染都方便了。不过要说一句，如果追求极致性能，默认游标确实比DictCursor快一丢丢，因为字典构建有开销。数据量在万级以下，这个差异基本可以忽略，可读性收益远大于性能损耗。

4.4 更新与删除的安全写法

更新和删除最核心的是WHERE条件要写明白，别一不留神全表更新。示例：

import pymysql conn = pymysql.connect( host='127.0.0.1', user='demo_user', password='your_password', database='demo_db', charset='utf8mb4' ) cursor = conn.cursor() # 更新 sql_update = "UPDATE users SET email = %s WHERE username = %s" cursor.execute(sql_update, ('new_email@example.com', 'alice')) # 删除 sql_delete = "DELETE FROM users WHERE username = %s" cursor.execute(sql_delete, ('alice',)) # 事务提交 conn.commit() print("影响行数：", cursor.rowcount) conn.close()

cursor.rowcount返回受影响的行数，这个值对业务判断很有用。比如，如果rowcount == 0，说明更新条件没匹配到任何记录，可能是数据已被删除或条件错误，业务上需要区分处理。

4.5 处理NULL值和类型转换

从MySQL取回NULL时，Python得到的是None，这点和JSON的null对应，但有些场景需要特殊处理：

email = row['email'] if email is None: email = '' # 业务层面把None转成空字符串

插入数据时也一样，Python的None会被正确存成MySQL的NULL。只是要注意，如果字段定义了NOT NULL，传None进去就会报错。数据库约束和代码逻辑要配合好，这点经验对减少线上事故非常有用。

5. 事务管理：保证数据一致性的核心机制

5.1 什么是事务与为什么需要它

事务是一组SQL操作的集合，要么全部成功，要么全部失败。经典场景是转账：A账户扣除100块，B账户增加100块。如果扣款成功但入账失败，没有事务保护就会出现资金凭空消失。

MySQL的InnoDB引擎是支持事务的，我们建表时指定的ENGINE=InnoDB就是为事务准备的。

5.2 PyMySQL中事务控制的正确姿势

PyMySQL的事务控制非常简单：

import pymysql conn = pymysql.connect( host='127.0.0.1', user='demo_user', password='your_password', database='demo_db', charset='utf8mb4' ) cursor = conn.cursor() try: # 关闭自动提交，开启一个事务 conn.autocommit(False) cursor.execute( "UPDATE users SET age = age + 1 WHERE username = %s", ('user_1',) ) cursor.execute( "UPDATE users SET age = age - 1 WHERE username = %s", ('user_2',) ) # 所有SQL执行成功，提交 conn.commit() print("事务提交成功") except Exception as e: # 任意一步出错，回滚到事务开始前的状态 conn.rollback() print("事务回滚：", e) finally: conn.close()

这段代码的关键点在于try...except...rollback的结构：所有SQL先缓存着不落库，直到commit()才真正执行；任何一个SQL报错都会跳到except，执行rollback()撤销前面已经执行的操作。

要不要在finally里再关闭连接？连接关闭时会自动回滚未提交的事务，所以即使忘记rollback()，关闭连接也不会造成脏数据提交。不过显式地写清楚提交回滚逻辑是更好的习惯。

5.3 自动提交设置：什么时候开什么时候关

很多初学朋友喜欢设置autocommit=True，因为不用手动commit()，省心。但实际业务中我劝你谨慎。

开启自动提交适合这些场景：

• 纯查询逻辑，没有写操作
• 非关键数据的写入，丢了也没太大关系
• 日志写入、统计上报类的数据

必须关闭自动提交的场景：

• 涉及多张表的写入操作
• 需要保持数据强一致性的业务（订单、支付、库存）
• 批量数据导入，中间失败要能整体回滚

还有一种说法是单条INSERT用不用事务都一样，其实不然。InnoDB即使单条INSERT也涉及事务，只是自动提交把每条SQL包成了独立事务。关闭自动提交批量插入时，最大的收益是减少了写日志（redo log）的同步次数，这也是批量操作性能提升的原因之一。

6. 进阶技巧：安全、性能与易用性提升

6.1 参数化查询：彻底告别SQL注入

SQL注入是非常危险的漏洞。举例说明：

# 危险写法，千万别这么干 username = input("请输入用户名：") sql = f"SELECT * FROM users WHERE username = '{username}'" cursor.execute(sql)

如果用户输入' OR '1'='1' --，拼出来的SQL就变成了：

SELECT * FROM users WHERE username = '' OR '1'='1' -- '

这个查询会把整张表的数据都带出来，用户输入被当成了代码执行。

正确写法是用占位符：

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

PyMySQL会把username当纯数据处理，永远不参与SQL语法解析。这是底线操作，任何动态拼接SQL的行为都必须禁止。

6.2 游标类型与遍历效率

前面提到DictCursor，PyMySQL游标类型其实有几种，根据场景选择：

SSCursor的机制不一样：普通游标是一次性把所有结果拉到客户端，SSCursor是在服务端保留游标，每取一条拉一条。几十万行数据用普通游标可能直接撑爆内存，换SSCursor就能稳定跑完。

import pymysql from pymysql.cursors import SSCursor conn = pymysql.connect( host='127.0.0.1', user='demo_user', password='your_password', database='demo_db', charset='utf8mb4', cursorclass=SSCursor ) cursor = conn.cursor() cursor.execute("SELECT id, username FROM users") # 此时不会一次性拉回所有数据 for row in cursor: # 逐行处理，内存占用恒定 print(row) conn.close()

有一点需要注意：使用SSCursor时，同一个连接上不能开着第二个游标执行别的查询，因为服务端游标是独占的。用完当前游标再开新的，否则会报错。

6.3 连接池：为什么你的程序老卡在连接上

每次请求都新建连接、用完关闭，并发一高，MySQL就会报Too many connections。连接池就是为了解决这个问题：预先建立一批连接，用的时候取，用完归还，循环利用。

PyMySQL本身没有连接池实现，但我们可以用DBUtils的PooledDB来做：

pip install DBUtils

from dbutils.pooled_db import PooledDB import pymysql pool = PooledDB( creator=pymysql, maxconnections=6, mincached=2, maxcached=5, blocking=True, host='127.0.0.1', user='demo_user', password='your_password', database='demo_db', charset='utf8mb4' ) # 从连接池拿连接 conn = pool.connection() cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM users") print(cursor.fetchone()) conn.close() # 注意：这里是归还连接，不是真正关闭

maxconnections=6表示池子最多持有6个连接，高峰期第七个请求会等待（blocking=True）直到有连接还回来。这比每次新建连接省下大量握手认证的时间。压测数据在我的机器上，连接池方式比每次新建连接吞吐量能提升大约3倍，延迟也稳定得多。

6.4 常见编码坑：emoji和中文乱码

MySQL的中文乱码属于经典问题，本质是字符集没对齐。三个层面需要一致：

• 数据库表的字符集
• 连接字符串的字符集
• 数据的实际编码

PyMySQL连接时指定charset='utf8mb4''，建表时也指定DEFAULT CHARSET=utf8mb4，基本就能解决。如果还乱码，检查MySQL服务端配置文件my.cnf里的默认字符集：

[client] default-character-set = utf8mb4 [mysql] default-character-set = utf8mb4 [mysqld] character-set-server = utf8mb4 collation-server = utf8mb4_unicode_ci

改完重启MySQL。还有一层注意：Python脚本文件本身要保存为UTF-8编码，编辑器右下角看一下编码格式，别让源文件编码和数据库编码打架。

7. 常见问题速查与避坑经验

7.1 踩过的坑和对应解决办法

整理一张速查表，都是实战中容易碰到且搜索频率高的问题：

7.2 唯一键冲突的高效处理

插入数据时经常遇到唯一键冲突。比如用户名已经存在，直接插入会报错。有两种处理思路：

第一种，先查再插，多一次网络往返，且有并发窗口风险：

cursor.execute("SELECT id FROM users WHERE username = %s", ('alice',)) if not cursor.fetchone(): cursor.execute("INSERT INTO users (username, email) VALUES (%s, %s)", ('alice', 'alice@example.com'))

第二种，直接INSERT ... ON DUPLICATE KEY UPDATE，一条SQL解决：

sql = """ INSERT INTO users (username, email, age) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE email = VALUES(email), age = VALUES(age) """ cursor.execute(sql, ('alice', 'alice@example.com', 30)) conn.commit()

第二种方案性能更好，语义也清晰：插入成功就插入，如果唯一键已存在，就更新指定字段。这个模式在同步数据的场景里几乎是标配。

7.3 经验之谈：排查问题的顺序

遇到数据库问题，我的排查顺序固定是：MySQL服务 → 网络连通 → 账号权限 → 字符集 → SQL语句。不要上来就怀疑PyMySQL有bug，这个库很成熟，99%的问题出在配置和环境上。

有一次排查线上问题，代码跑得好好的突然开始报连接超时，最后发现是数据库服务器磁盘满了，连接建立变慢。这种问题从应用层看毫无头绪，但排查思路要广，CPU、内存、磁盘、网络、连接数，五个维度都要过一遍。

8. 动手实践：一个完整的小示例

写一个完整的用户管理小例子，把前面的知识点串起来：连接、查询、插入、事务、上下文管理。

import pymysql from pymysql.cursors import DictCursor class UserDB: def __init__(self, host, user, password, database): self.conn = pymysql.connect( host=host, user=user, password=password, database=database, charset='utf8mb4', cursorclass=DictCursor, autocommit=False ) def get_user(self, username): with self.conn.cursor() as cursor: cursor.execute( "SELECT id, username, email, age FROM users WHERE username = %s", (username,) ) return cursor.fetchone() def add_user(self, username, email, age): with self.conn.cursor() as cursor: try: cursor.execute( "INSERT INTO users (username, email, age) VALUES (%s, %s, %s)", (username, email, age) ) self.conn.commit() return cursor.lastrowid except pymysql.IntegrityError: self.conn.rollback() raise ValueError(f"用户名 {username} 已存在") def update_user(self, username, email=None, age=None): with self.conn.cursor() as cursor: if email: cursor.execute( "UPDATE users SET email = %s WHERE username = %s", (email, username) ) if age is not None: cursor.execute( "UPDATE users SET age = %s WHERE username = %s", (age, username) ) self.conn.commit() return cursor.rowcount def list_users(self, limit=10): with self.conn.cursor() as cursor: cursor.execute("SELECT id, username, email, age FROM users LIMIT %s", (limit,)) return cursor.fetchall() def close(self): self.conn.close() if __name__ == '__main__': db = UserDB('127.0.0.1', 'demo_user', 'your_password', 'demo_db') try: new_id = db.add_user('bob', 'bob@example.com', 25) print("新增用户ID：", new_id) user = db.get_user('bob') print("查询结果：", user) db.update_user('bob', age=26) print("更新后：", db.get_user('bob')) for u in db.list_users(): print(u) finally: db.close()

这个类把连接、操作、提交、回滚、异常处理都封装起来了，实际项目里可以在这个基础上扩展更多方法，或者改造成支持连接池的版本。

9. 实操心得与小技巧

整篇内容到这里，基于我自己长期折腾PyMySQL的经验，最后再分享几个小技巧。

第一个小技巧：排查性能问题时，打开MySQL的慢查询日志，看看哪些SQL消耗了大量时间。很多时候瓶颈不在PyMySQL，而是SQL本身没有走索引，或者查询方式有问题。

第二个建议：代码里所有数据库连接参数不要硬编码。用环境变量或者配置文件管理，做到代码和配置分离。这样从开发环境切到测试、生产环境，只需要改配置，不用动代码逻辑。

第三个经验：在写批量插入、更新这类操作时，先在小数据集上跑通逻辑，再上全量。数据库操作不像普通计算能随意撤销，一旦执行范围有误，影响的可能是全表数据。

第四个技巧，也是我反复提到的：任何时候拿不准操作的影响范围，先写SELECT确认，再执行UPDATE或DELETE。不上生产环境前，每个写操作面前都要有敬畏心。

PyMySQL这个库本身不算复杂，真正的复杂度在如何把连接管理、事务控制、性能优化和业务逻辑组合好。希望这篇文章能帮你少走弯路，把更多时间花在业务本身，而不是在连接报错里反复挣扎。