Python3操作MySQL全指南：驱动选择、连接与事务实践

简介：面向已经具备一定 Python 基础的开发者，详细讲解基于 Python 3.7 与 PyMySQL 0.9.3 连接 MySQL 数据库，并完成数据的查询、新增、修改与删除等常用操作。内容以一个封装好的数据库连接类为线索，先说明连接参数的配置方式，包括主机地址、端口、用户名、密码、数据库名以及字符集；随后分别介绍用于变更数据和查询数据的两个方法。变更方法会先判断传入的 SQL 是否为空，再建立连接、创建游标、执行语句并提交事务；查询方法则负责执行查询命令并获取全部返回结果，两者都包含异常捕获和连接关闭的处理。整份资源为单个 PDF 文件，大小约五十 KB，便于快速下载和离线阅读；已有 2395 人学习使用，既适合刚开始接触 Python 数据库开发的初学者了解整体流程，也适合需要封装数据库操作类的开发者参考。通过文档中的完整代码与调用示例，可以快速掌握连接配置、SQL 执行、多数据库切换以及错误排查等实用技巧，并将这些实现直接应用到实际项目中。

1. 为什么是 Python3 + MySQL，以及你真正要装的驱动

如果你在一台新机器上准备用 Python3 操作 MySQL，先别急着开始写代码，第一步大概率会栽在驱动安装上。网上大量教程会让你直接pip install mysqlclient，但 Windows 上这个包经常因为缺少编译环境而报错，Linux 上又要求你预先装libmysqlclient-dev。实际上，对绝大多数应用场景，更省事的选择是 PyMySQL：它是纯 Python 实现，不依赖本地 C 库，pip install pymysql一行搞定，连 MySQL 8.0 的caching_sha2_password认证也支持。这篇文章围绕「python+mysql数据库连接及操作」这个最常被搜索的标题，把从建立连接、增删改查、事务控制到连接池、常见异常的全部路径走一遍，适合刚入门 Python 的开发者，也适合那些想搞清楚游标、事务边界、参数化查询细节的熟手。下面提到的每个参数都会解释含义，每段代码都能直接跑。

2. Python3 连接 MySQL 的最小可用代码：连接参数与游标选择

2.1 先搞定驱动安装：pip 装 PyMySQL 还是 mysql-connector-python

操作 MySQL 的 Python 驱动主要有三个选择：mysqlclient、MySQL Connector/Python和PyMySQL。其中mysqlclient是MySQLdb的衍生版，C 扩展编译，性能好，但安装时对系统依赖要求高；mysql-connector-python是官方驱动，功能最完整，但包体较大，且早期版本在某些场景下性能和 PyMySQL 差不多，文档风格也偏官方；PyMySQL是纯 Python 实现，兼容MySQLdb的 API，几乎不需要额外依赖，是目前社区里最常被推荐的方案。

我一般会优先选 PyMySQL，理由有三个：第一，它在虚拟环境里安装不会触发编译错误；第二，它支持MySQLdb风格的调用方式，以后要换mysqlclient，代码迁移成本几乎为零；第三，它对 Python3.6 以上的现代语法支持良好，配合pymysql.cursors.DictCursor可以直接得到字典格式的结果。安装命令如下。

pip install pymysql # 如果使用 Anaconda，也可以使用 conda install pymysql

装上之后，在 Python 交互式环境里执行import pymysql，如果没有报错，驱动就绪。除了驱动，你还需要一个能连的 MySQL 服务。如果你正在看这篇，大概率还在折腾mysql安装配置教程，最简单的本地验证方式是先通过命令行工具mysql -u root -p确认服务已启动，再回来写 Python 代码。

2.2 建立连接：host port user password database 与 charset

连接 MySQL 的入口是pymysql.connect()。它的参数很多，但核心就几个：host是数据库地址，本地用127.0.0.1，不要用localhost，因为在某些系统上localhost会走 Unix socket，而 Python 驱动默认走 TCP；port默认3306；user和password对应 MySQL 账号；database是要操作的库名；charset建议显式指定utf8mb4，这样才能完整保存 Emoji 和生僻汉字，而不是utf8。

下面是最小连接代码，包含一个简单的连通性测试。

import pymysql connection = pymysql.connect( host='127.0.0.1', port=3306, user='root', password='your_password', database='test_db', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) with connection.cursor() as cursor: cursor.execute("SELECT VERSION()") result = cursor.fetchone() print(result) connection.close()

逻辑说明：cursorclass指定返回结果类型，DictCursor会返回字典，字段名作为 key，适合大多数业务代码。with connection.cursor()语句会自动关闭游标，但不会自动关闭连接，所以脚本末尾仍需connection.close()。参数说明：charset的值不要传utf-8，中间没有连字符；password如果包含特殊字符，注意转义问题。另外，connect()是阻塞式的，连接超时默认是 10 秒，受connect_timeout参数控制，在生产环境建议显式设置一个合理值。

2.3 游标类型与 fetch 结果：cursor、buffered 与原生字典

游标是数据库驱动里负责执行语句和获取结果的对象。PyMySQL 默认的Cursor是「非缓冲」的，意思是执行execute()后结果集由 MySQL 服务端持续发送，客户端需要主动fetch。这个模式的优点是不一次性占用内存，缺点是当你还没fetch完就执行第二条查询时，会抛raise an error。更早踩过这个坑的人会在连接参数里加一个cursorclass=pymysql.cursors.SSCursor，也就是「无缓冲游标」，但那是为了流式读大结果集时用的，普通查询不要用。

实际开发中，我更推荐直接使用DictCursor，但要注意它与SSCursor的组合会变成SSDictCursor，这个类型同样存在「未取完不能开新查询」的限制。为了不踩这个坑，最简单的办法是，在需要立刻执行多条语句的场景下，使用conn.begin()配合事务，或者在一条查询后用fetchall()把数据取完。下表列出了常见游标类型：

需要记住的是，如果只用默认游标，fetchone()返回值是一个元祖，比如('8.0.36',)，而DictCursor得到的是{'VERSION()': '8.0.36'}。在写增删改查代码前，先把这个差异搞清楚，能避免后来所有代码里row[0]还是row['id']的纠结。

3. 数据库连接及操作的核心：增删改查与参数化查询

3.1 用 execute() 执行 INSERT/UPDATE/DELETE 的正确姿势

连接建好，游标拿到手，接下来是核心操作。先建一张测试表，假设要做一个简单的用户表：

CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50) NOT NULL, age INT DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

插入单条记录的 Python 代码：

import pymysql connection = pymysql.connect( host='127.0.0.1', user='root', password='your_password', database='test_db', charset='utf8mb4', autocommit=True ) with connection.cursor() as cursor: sql = "INSERT INTO users (name, age) VALUES (%s, %s)" cursor.execute(sql, ("张三", 25)) last_id = cursor.lastrowid print("新增ID:", last_id) sql_update = "UPDATE users SET age = %s WHERE name = %s" cursor.execute(sql_update, (26, "张三")) sql_delete = "DELETE FROM users WHERE name = %s" cursor.execute(sql_delete, ("李四",)) connection.close()

逻辑说明：%s是占位符，不管字段是什么类型，统一用%s，第二个参数传入元组。cursor.lastrowid拿到的是自增字段的 ID，只在 INSERT 后有效。UPDATE和DELETE的execute()返回的是受影响行数，如果你想知道删了几条，可以rowcount = cursor.rowcount。参数说明：即使DELETE只有一个条件值，也要写成("李四",)加逗号，表示一个元组，否则会被当成字符串序列断开。

注意上面的代码设置了autocommit=True。如果不设置，那么每次 DML 操作后必须手动connection.commit()，否则数据不会真正落到磁盘。对初学者来说，autocommit=True能避免「明明执行了但数据没变」的困惑；但对事务要严格控制的业务系统，应该关掉它，改成显式提交。

3.2 事务要么全做要么全不做：commit 与 rollback 的边界

事务是 MySQL 里最容易理解但又最容易写错的部分。PyMySQL 的默认行为是「开启一个隐式事务」，也就是说执行第一条 DML 语句后，事务自动开始，直到你调用commit()或rollback()结束。看一个典型的转账场景：

import pymysql connection = pymysql.connect( host='127.0.0.1', user='root', password='your_password', database='test_db', charset='utf8mb4', autocommit=False ) try: with connection.cursor() as cursor: cursor.execute("UPDATE users SET age = age - 1 WHERE name = '张三'") cursor.execute("UPDATE users SET age = age + 1 WHERE name = '王五'") # 两条语句都成功，才提交 connection.commit() print("事务已提交") except Exception as e: # 任何一条失败，回滚 connection.rollback() print("事务已回滚:", e) finally: connection.close()

逻辑说明：autocommit=False时，execute()后的数据只在当前会话可见，其他连接看不到，只有commit()后才会全局生效。rollback()会将本次事务里所有未提交的变更全部撤掉。这段代码里，如果第二条UPDATE因为字段约束失败，整个事务回滚，第一条更新也不会生效。参数说明：autocommit是连接级别的参数，不是每次操作级别。如果你在同一个连接里要混合「自动提交的查询」和「需要事务的批量写入」，最常见的做法是保持autocommit=False，然后在你认为每条独立语句执行完后手动commit()。

需要特别提醒的是，PyMySQL 中BEGIN/COMMIT是可以手动发出去的，但在事务进行中执行cursor.execute("COMMIT")会导致连接状态混乱，因为驱动内部有自己的事务状态标记。正确做法是永远使用connection.commit()和connection.rollback()。

3.3 查询结果处理：fetchone、fetchmany、fetchall 与内存控制

SELECT 查询返回的结果集，驱动提供了三种取出方式：fetchone()取一行，fetchmany(n)取 n 行，fetchall()取全部。对大数据量查询，fetchall()会把结果全部载入 Python 列表，非常消耗内存。看下面对比代码：

with connection.cursor() as cursor: cursor.execute("SELECT id, name, age FROM users") # 方法1：一次取一条，适合快速试探 first_row = cursor.fetchone() print("第一条:", first_row) # 方法2：一次取固定条数，适合分页 rows_100 = cursor.fetchmany(100) print("接下来100条:", len(rows_100)) # 方法3：全取，适合小结果集 all_rows = cursor.fetchall() print("总数:", len(all_rows))

逻辑说明：游标是有状态的，fetchone()后会向下移动，所以三种方法是串行消费同一结果集，混用时不要搞错位置。如果你只想统计行数，用SELECT COUNT(*) AS cnt FROM users而不是fetchall()。参数说明：fetchmany(100)的 100 不是分页参数，它只是每次从网络缓冲区取多少条，如果底层结果只剩 30 条，返回 30 条，不会报错也不会补空行。

在数据量大到数十万行的场景，fetchall()会让 Python 进程的内存占用直线上升。更稳妥的方式是用SSCursor做流式读取，但前面说过，它不允许在未消费完前执行新查询。所以实践中，我会先评估结果集大小——通常WHERE条件能把结果压到几千行以内，直接用fetchall()最省事；超过这个量，就要考虑分批查询，也就是用LIMIT和OFFSET做翻页，而不是靠驱动缓冲。

3.4 一个容易踩的坑：拼接 SQL 与注入防护

这是数据库连接及操作中最常见的安全问题。很多初学者会写出这样的代码：

name = input("输入用户名: ") sql = f"SELECT * FROM users WHERE name = '{name}'" cursor.execute(sql)

问题是，如果输入' OR 1=1 --，拼接出来的 SQL 就变成了WHERE name = '' OR 1=1 -- '，单引号被闭合，--注释掉后面内容，整张表会被查出来。正确做法就是前面一直使用的%s占位符：

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

逻辑说明：PyMySQL 在收到占位符形式的 SQL 后，会把参数转义成一个合法的字符串字面量，也就是把'替换成\'，从源头杜绝注入。参数说明：占位符%s的位置不要加引号，加了引号就成了字符串拼接，%(name)s这种命名占位符风格在 PyMySQL 里兼容，但最好统一用%s。另外，cursor.execute()的第二个参数必须是序列或字典，传字符串会被当成单个字符序列，导致参数数量对不上。

如果要执行多条 SQL，cursor.executemany()的用法是这样的：

sql = "INSERT INTO users (name, age) VALUES (%s, %s)" data = [("a", 1), ("b", 2), ("c", 3)] cursor.executemany(sql, data)

它会批量执行并返回受影响的总行数，性能远优于循环单个execute()，这在批量导入时是必用选项。

4. 连接池、异常处理与 MySQL 8.0 的认证坑

4.1 MySQL 8.0 默认认证插件导致连接失败怎么处理

如果你用的是mysql安装教程8.0装出来的数据库，然后用 PyMySQL 连接，大概率会碰上Authentication plugin 'caching_sha2_password' cannot be loaded类似报错。这是因为 MySQL 8.0 默认的认证插件是caching_sha2_password，而有些旧驱动或旧版本的 PyMySQL 不支持。

PyMySQL 从 0.9.3 开始支持该插件，所以第一个排查方向是把 PyMySQL 升级到最新版：

pip install -U pymysql

如果升级后仍然报错，常见做法是在 MySQL 端把账号的认证插件改写为mysql_native_password：

ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY '你的密码'; FLUSH PRIVILEGES;

但要说明，mysql_native_password是旧算法，在 MySQL 8.0 中已经标记为废弃，官方推荐逐步迁移到caching_sha2_password。我更推荐的做法是：确保 PyMySQL 版本足够新，并且连接参数里显式设置charset="utf8mb4"，避免 SSL 相关干扰。如果你的 MySQL 服务端开启了 SSL 但客户端没有证书，报错可能不是认证插件，而是Access denied for user，那就需要在connect()里设置ssl_disabled=True试试看。

4.2 用 DBUtils 实现线程安全的连接池

每个业务请求都新建一个 MySQL 连接，在高并发下会立刻把数据库连接数打满，因为建立 TCP 连接、认证、分配资源一套流程开销不小。常驻进程应用（比如 FastAPI、Flask 长驻 worker）应该使用连接池复用连接。PyMySQL 本身不带连接池，最常用的组合是DBUtils.PooledDB。

安装和基础配置如下：

# pip install DBUtils from dbutils.pooled_db import PooledDB import pymysql pool = PooledDB( creator=pymysql, maxconnections=10, mincached=2, maxcached=5, blocking=True, host='127.0.0.1', user='root', password='your_password', database='test_db', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) # 从池中取一个连接 connection = pool.connection()

逻辑说明：creator=pymysql表示池化的是 PyMySQL 连接。maxconnections=10是池的最大连接数，mincached=2是启动时就保持 2 个空闲连接，maxcached=5是池中最多保留 5 个空闲连接。blocking=True表示当连接被借完时，新的调用会阻塞等待，而不是直接抛异常。参数说明：取出的connection用完后调close()不是真正关闭连接，而是把连接归还给池。如果你在finally里不 close，池中的连接会被耗尽。

关于连接池还有一点要注意：连接池里的连接可能因为 MySQL 服务端wait_timeout而被服务端断开，此时从池中拿到的连接是「死的」。所以真正使用连接之前，建议先执行一个廉价查询，比如cursor.execute("SELECT 1")，失败则丢弃这个连接并重试。DBUtils 的PooledDB参数ping可以控制连接回收时检测，设置ping=1会有一定的探测开销，但能避免上述问题。

4.3 关键异常与排查路径：InterfaceError、OperationalError、DataError

Python 数据库开发里，你迟早会遇到这三类异常，它们的处理方式完全不同。

pymysql.err.InterfaceError通常表示连接已经关闭或不可用。比如在with connection.cursor()外面使用同一个连接，但连接被手动 close 了；或者连接池里拿到的连接被服务端断开。排查路径：检查代码路径是否在finally里 close 了连接；检查查询耗时是否超过了 MySQL 的wait_timeout；加日志打印connection.open属性，False说明连接已断。

pymysql.err.OperationalError是最常见的数据库操作异常，错误码通常是1045（拒绝访问）、1049（未知数据库）、2003（无法连接服务器）、2013（查询期间丢失连接）。2003 出现时，先确认host/port是否可通，Linux 上还要检查防火墙；2013 通常和大查询或超时有关，MySQL 端要关注net_write_timeout和max_allowed_packet。特别是导入大文本或 blob 时，如果写入超过max_allowed_packet，报错会是Packet too large，处理方式是调整 MySQL 变量：

SET GLOBAL max_allowed_packet = 64 * 1024 * 1024;

pymysql.err.DataError是因为数据值不符合字段定义，比如插入字符串超过VARCHAR长度，或者给INT字段传了超大数值。遇到这种异常，不要光看 Python 堆栈，要把 SQL 和参数一起打出来，用 MySQL 客户端跑一遍同样的语句，错误信息更直接。

下面是一个带异常处理的完整连接池使用模板：

from dbutils.pooled_db import PooledDB import pymysql pool = PooledDB( creator=pymysql, maxconnections=5, mincached=1, maxcached=3, blocking=True, host='127.0.0.1', user='root', password='your_password', database='test_db', charset='utf8mb4' ) def query_one(sql, args=None): conn = pool.connection() try: with conn.cursor() as cur: cur.execute(sql, args) return cur.fetchone() except pymysql.err.OperationalError as e: print("操作失败，错误码:", e.args[0], "信息:", e.args[1]) return None finally: conn.close() print(query_one("SELECT id, name FROM users WHERE id = %s", (1,)))

逻辑说明：e.args[0]是 MySQL 错误码，e.args[1]是错误文本，这两个字段在排查问题时非常有用。finally里conn.close()把连接还回池而不是直接销毁。参数说明：连接池参数不是越多越好，maxconnections一般设置为应用服务器 CPU 核数两倍左右，具体要压测。

5. 一个技巧：用上下文管理器把连接写干净，再验证索引与排序

最后一章留给一个可以长期沿用的编码习惯：利用contextlib.contextmanager把连接和游标的生命周期包装起来，让业务代码里不再出现 try/finally 嵌套。同时，用mysql排序和mysql创建索引这两个高频场景来验证写出来的代码不是「只能跑通」，而是「能写出有效查询」。

常见的做法是写一个get_cursor上下文管理器，负责从连接池获取连接、提供游标、自动提交或回滚、归还连接。

from contextlib import contextmanager from dbutils.pooled_db import PooledDB import pymysql pool = PooledDB( creator=pymysql, maxconnections=5, mincached=1, maxcached=3, blocking=True, host='127.0.0.1', user='root', password='your_password', database='test_db', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) @contextmanager def db_cursor(commit=False): conn = pool.connection() try: with conn.cursor() as cur: yield cur if commit: conn.commit() except Exception: conn.rollback() raise finally: conn.close()

使用它，业务代码变得非常简洁：

with db_cursor() as cur: cur.execute("SELECT id, name, age FROM users ORDER BY age DESC, id ASC LIMIT 10") top_10 = cur.fetchall() for row in top_10: print(row["id"], row["name"], row["age"])

这段代码中ORDER BY age DESC, id ASC表示先按 age 倒序，age 相同再按 id 正序。很多刚接触mysql排序的人会忽略排序字段的索引优化。当表数据量大时，ORDER BY age会触发 filesort，如果age上有索引，MySQL 就能直接按索引顺序读取，避免临时文件和额外的排序操作。验证一个查询是否走了索引，可以在 MySQL 命令行或客户端工具里执行：

EXPLAIN SELECT id, name, age FROM users ORDER BY age DESC LIMIT 10;

看到Extra列有Using index condition或Using where是好事，如果出现Using filesort，说明需要检查索引设计。创建索引的命令是：

CREATE INDEX idx_users_age ON users(age);

创建后再跑一次EXPLAIN，会发现Using filesort消失，这就是索引带来的实际收益。在 Python 代码里，你只负责写ORDER BY和LIMIT，索引是否命中由 MySQL 优化器决定，而优化器依赖于统计信息，所以定期ANALYZE TABLE users也很重要。

再把注意力从验证拉回到连接管理。使用db_cursor有个明确边界：你可以在with块里做多次execute()，但默认commit=False，也就是所有改动只在当前事务内，只有显式传commit=True才整体提交。这种设计避免了「每条 DML 后忘记 commit」，也避免了大事务里意外提前提交。在调用处可以这样用：

with db_cursor(commit=True) as cur: cur.execute("INSERT INTO users (name, age) VALUES (%s, %s)", ("测试", 18))

回到查询验证：上面top_10拿到的是一个列表，里面是字典。fetchall()之后游标已经走到结果集末尾，如果你再执行同一条新 SQL 也是允许的，因为非缓冲游标在fetchall()后已结束。刚才提到的大结果集场景，改成fetchmany(1000)循环读取，同样可以在with db_cursor()里完成，唯一要注意的是不要在游标未消费完时调用同一个连接上的另一个execute()。如果你确实要在一个连接里执行多条独立查询，等到每一条都fetchall()再执行下一条，就不会碰到Commands out of sync的错误提示。

本文还有配套的精品资源，点击获取