Python通过PyMySQL连接MySQL的详细教程(新手必看)
作者:菩提风
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 连接参数背后的逻辑
既然要深入,把常用连接参数讲透,以后用什么都受益:
| 参数 | 默认值 | 说明与建议 |
|---|---|---|
host | localhost | 数据库地址,线上用内网IP或域名 |
port | 3306 | MySQL服务端口 |
user | 无 | 用户名,建议按权限拆分 |
password | 无 | 密码,代码里别明文写死,环境变量或密钥管理 |
database | 无 | 连接的默认数据库 |
charset | utf8mb4 | 字符集,统一设置避免乱码 |
connect_timeout | 10 | 连接超时秒数,默认10秒在某些网络环境下不够,按需调整 |
cursorclass | 默认游标 | 可以指定 DictCursor 等,后面会讲 |
autocommit | False | 是否自动提交事务,默认关闭是MySQL的保守设定 |
max_allowed_packet | 67108864 | 数据包大小上限,插入大数据时需要调大 |
有一个容易被忽略的点: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游标类型其实有几种,根据场景选择:
| 游标类型 | 返回格式 | 适用场景 |
|---|---|---|
默认 Cursor | 元组 | 批量数据处理,性能优先 |
DictCursor | 字典 | 业务代码,可读性优先 |
SSCursor | 元组,服务端游标 | 超大结果集,边取边处理 |
SSDictCursor | 字典,服务端游标 | 超大结果集且需要字典格式 |
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 踩过的坑和对应解决办法
整理一张速查表,都是实战中容易碰到且搜索频率高的问题:
| 报错或现象 | 原因 | 解决方案 |
|---|---|---|
Can't connect to MySQL server on 'localhost' | MySQL服务未启动,或IPv6解析问题 | 确认服务状态,host改用 127.0.0.1 |
Access denied for user 'xxx'@'localhost' | 账号密码错误,或权限不足 | 检查密码,用 GRANT 正确授权 |
Unknown database 'xxx' | 数据库不存在或未创建 | 先创建数据库,检查名称拼写 |
Table 'xxx' doesn't exist | 表不存在,或当前连接的数据库不正确 | 检查 database 参数,确认表名 |
Duplicate entry 'xxx' for key 'uk_username' | 唯一索引冲突 | 先用 SELECT 判断是否存在,或使用 INSERT ... ON DUPLICATE KEY UPDATE |
Incorrect string value: '\xF0\x9F\x98\x80' | 字符集不支持emoji | 表结构、连接参数统一改为 utf8mb4 |
Lost connection to MySQL server during query | 数据包超限,或连接空闲太久被断开 | 调整 max_allowed_packet ,或设置 wait_timeout |
MySQL server has gone away | 连接长时间无活动被服务端断开 | 使用连接池保活,或重连机制 |
| 数据查询结果显示为乱码 | 字符集各环节不一致 | 统一 utf8mb4 ,检查文件编码和连接参数 |
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这个库本身不算复杂,真正的复杂度在如何把连接管理、事务控制、性能优化和业务逻辑组合好。希望这篇文章能帮你少走弯路,把更多时间花在业务本身,而不是在连接报错里反复挣扎。
以上就是Python通过PyMySQL连接MySQL的详细教程(新手必看)的详细内容,更多关于Python PyMySQL连接MySQL的资料请关注脚本之家其它相关文章!
