python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python PyMySQL连接MySQL

Python通过PyMySQL连接MySQL的详细教程(新手必看)

作者:菩提风

本文系统讲解了Python通过PyMySQL连接MySQL的核心技术,从安装配置到增删改查,从事务管理到连接池优化,手把手教你避开常见坑,学会参数化查询防SQL注入,用executemany提升10倍插入效率,轻松搞定数据库操作

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;

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

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()

注意几个细节:

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方法:

它们的使用区别要搞清楚,特别是内存消耗。数据量小用 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的资料请关注脚本之家其它相关文章!

您可能感兴趣的文章:
阅读全文