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认证也支持。这篇文章围绕「pythonmysql数据库连接及操作」这个最常被搜索的标题把从建立连接、增删改查、事务控制到连接池、常见异常的全部路径走一遍适合刚入门 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 驱动默认走 TCPport默认3306user和password对应 MySQL 账号database是要操作的库名charset建议显式指定utf8mb4这样才能完整保存 Emoji 和生僻汉字而不是utf8。下面是最小连接代码包含一个简单的连通性测试。import pymysql connection pymysql.connect( host127.0.0.1, port3306, userroot, passwordyour_password, databasetest_db, charsetutf8mb4, cursorclasspymysql.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。更早踩过这个坑的人会在连接参数里加一个cursorclasspymysql.cursors.SSCursor也就是「无缓冲游标」但那是为了流式读大结果集时用的普通查询不要用。实际开发中我更推荐直接使用DictCursor但要注意它与SSCursor的组合会变成SSDictCursor这个类型同样存在「未取完不能开新查询」的限制。为了不踩这个坑最简单的办法是在需要立刻执行多条语句的场景下使用conn.begin()配合事务或者在一条查询后用fetchall()把数据取完。下表列出了常见游标类型游标类型返回结果缓冲方式适用场景Cursor元组非缓冲内存敏感、逐行处理DictCursor字典非缓冲业务代码、ORM 替代SSCursor元组流式超大结果集SSDictCursor字典流式超大结果集且要字段名需要记住的是如果只用默认游标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 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;插入单条记录的 Python 代码import pymysql connection pymysql.connect( host127.0.0.1, userroot, passwordyour_password, databasetest_db, charsetutf8mb4, autocommitTrue ) 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只有一个条件值也要写成(李四,)加逗号表示一个元组否则会被当成字符串序列断开。注意上面的代码设置了autocommitTrue。如果不设置那么每次 DML 操作后必须手动connection.commit()否则数据不会真正落到磁盘。对初学者来说autocommitTrue能避免「明明执行了但数据没变」的困惑但对事务要严格控制的业务系统应该关掉它改成显式提交。3.2 事务要么全做要么全不做commit 与 rollback 的边界事务是 MySQL 里最容易理解但又最容易写错的部分。PyMySQL 的默认行为是「开启一个隐式事务」也就是说执行第一条 DML 语句后事务自动开始直到你调用commit()或rollback()结束。看一个典型的转账场景import pymysql connection pymysql.connect( host127.0.0.1, userroot, passwordyour_password, databasetest_db, charsetutf8mb4, autocommitFalse ) 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()逻辑说明autocommitFalse时execute()后的数据只在当前会话可见其他连接看不到只有commit()后才会全局生效。rollback()会将本次事务里所有未提交的变更全部撤掉。这段代码里如果第二条UPDATE因为字段约束失败整个事务回滚第一条更新也不会生效。参数说明autocommit是连接级别的参数不是每次操作级别。如果你在同一个连接里要混合「自动提交的查询」和「需要事务的批量写入」最常见的做法是保持autocommitFalse然后在你认为每条独立语句执行完后手动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 fSELECT * FROM users WHERE name {name} cursor.execute(sql)问题是如果输入 OR 11 --拼接出来的 SQL 就变成了WHERE name OR 11 -- 单引号被闭合--注释掉后面内容整张表会被查出来。正确做法就是前面一直使用的%s占位符sql SELECT * FROM users WHERE name %s cursor.execute(sql, (name,))逻辑说明PyMySQL 在收到占位符形式的 SQL 后会把参数转义成一个合法的字符串字面量也就是把替换成\从源头杜绝注入。参数说明占位符%s的位置不要加引号加了引号就成了字符串拼接%(name)s这种命名占位符风格在 PyMySQL 里兼容但最好统一用%s。另外cursor.execute()的第二个参数必须是序列或字典传字符串会被当成单个字符序列导致参数数量对不上。如果要执行多条 SQLcursor.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_passwordALTER USER rootlocalhost IDENTIFIED WITH mysql_native_password BY 你的密码; FLUSH PRIVILEGES;但要说明mysql_native_password是旧算法在 MySQL 8.0 中已经标记为废弃官方推荐逐步迁移到caching_sha2_password。我更推荐的做法是确保 PyMySQL 版本足够新并且连接参数里显式设置charsetutf8mb4避免 SSL 相关干扰。如果你的 MySQL 服务端开启了 SSL 但客户端没有证书报错可能不是认证插件而是Access denied for user那就需要在connect()里设置ssl_disabledTrue试试看。4.2 用 DBUtils 实现线程安全的连接池每个业务请求都新建一个 MySQL 连接在高并发下会立刻把数据库连接数打满因为建立 TCP 连接、认证、分配资源一套流程开销不小。常驻进程应用比如 FastAPI、Flask 长驻 worker应该使用连接池复用连接。PyMySQL 本身不带连接池最常用的组合是DBUtils.PooledDB。安装和基础配置如下# pip install DBUtils from dbutils.pooled_db import PooledDB import pymysql pool PooledDB( creatorpymysql, maxconnections10, mincached2, maxcached5, blockingTrue, host127.0.0.1, userroot, passwordyour_password, databasetest_db, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor ) # 从池中取一个连接 connection pool.connection()逻辑说明creatorpymysql表示池化的是 PyMySQL 连接。maxconnections10是池的最大连接数mincached2是启动时就保持 2 个空闲连接maxcached5是池中最多保留 5 个空闲连接。blockingTrue表示当连接被借完时新的调用会阻塞等待而不是直接抛异常。参数说明取出的connection用完后调close()不是真正关闭连接而是把连接归还给池。如果你在finally里不 close池中的连接会被耗尽。关于连接池还有一点要注意连接池里的连接可能因为 MySQL 服务端wait_timeout而被服务端断开此时从池中拿到的连接是「死的」。所以真正使用连接之前建议先执行一个廉价查询比如cursor.execute(SELECT 1)失败则丢弃这个连接并重试。DBUtils 的PooledDB参数ping可以控制连接回收时检测设置ping1会有一定的探测开销但能避免上述问题。4.3 关键异常与排查路径InterfaceError、OperationalError、DataErrorPython 数据库开发里你迟早会遇到这三类异常它们的处理方式完全不同。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( creatorpymysql, maxconnections5, mincached1, maxcached3, blockingTrue, host127.0.0.1, userroot, passwordyour_password, databasetest_db, charsetutf8mb4 ) def query_one(sql, argsNone): 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( creatorpymysql, maxconnections5, mincached1, maxcached3, blockingTrue, host127.0.0.1, userroot, passwordyour_password, databasetest_db, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor ) contextmanager def db_cursor(commitFalse): 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()但默认commitFalse也就是所有改动只在当前事务内只有显式传commitTrue才整体提交。这种设计避免了「每条 DML 后忘记 commit」也避免了大事务里意外提前提交。在调用处可以这样用with db_cursor(commitTrue) 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的错误提示。本文还有配套的精品资源点击获取