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想连数据库的新手被各种连接报错折磨的排查者需要在脚本里批量处理数据的开发者以及想搞清楚事务和连接池原理的进阶玩家。我会把从安装到实战、从连接到优化、从常用操作到避坑经验都过一遍内容偏实操直接能抄。提示如果你用的是MariaDBPyMySQL也同样支持它在驱动层面的兼容性是没问题的。2. 准备工作环境安装与依赖梳理2.1 快速安装PyMySQL安装非常简单pip一把梭pip install pymysql如果你在多个Python环境间切换或者要用虚拟环境先激活再装别装错环境。装了多个Python版本的老手经常遇到“明明装了我却导入不了”的尴尬就是环境串了。验证是否装好进Python交互环境试一下import pymysql print(pymysql.__version__)能输出版本号就说明安装成功。我这边习惯在项目里用requirements.txt锁版本避免过阵子库更新导致行为变化pymysql1.1.02.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_userlocalhost IDENTIFIED BY your_password; GRANT SELECT, INSERT, UPDATE, DELETE ON demo_db.* TO demo_userlocalhost; 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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;这张表包含了自增主键、唯一索引、默认值时间戳足够覆盖大部分入门场景。建完表之后我们就能把注意力完全放到Python代码上了。3. 核心连接逻辑与关键参数3.1 最朴素的连接方式PyMySQL的connect方法是最常用的入口。来一段最小可用示例import pymysql conn pymysql.connect( host127.0.0.1, port3306, userdemo_user, passwordyour_password, databasedemo_db, charsetutf8mb4 ) 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::1MySQL默认监听IPv4就容易踩Cant connect to MySQL server on localhost的坑。用127.0.0.1能避免绝大多数这类问题。端口3306是MySQL默认端口如果你改了配置这里要跟着改。游标和连接用完必须关闭Python的finally块就是干这个的。不关的话连接会一直占着数据库连接数飙升最终服务拒绝新连接。3.2 连接参数背后的逻辑既然要深入把常用连接参数讲透以后用什么都受益参数默认值说明与建议hostlocalhost数据库地址线上用内网IP或域名port3306MySQL服务端口user无用户名建议按权限拆分password无密码代码里别明文写死环境变量或密钥管理database无连接的默认数据库charsetutf8mb4字符集统一设置避免乱码connect_timeout10连接超时秒数默认10秒在某些网络环境下不够按需调整cursorclass默认游标可以指定DictCursor等后面会讲autocommitFalse是否自动提交事务默认关闭是MySQL的保守设定max_allowed_packet67108864数据包大小上限插入大数据时需要调大有一个容易被忽略的点PyMySQL默认autocommitFalse也就是说你执行INSERT、UPDATE、DELETE之后必须手动conn.commit()否则数据不会真正落库。新手最容易在这里迷糊明明执行没报错一看表里数据没变就是因为没提交事务。这个行为是保护机制不是bug配合事务使用才是正确的姿势。3.3 使用with语法简化连接管理Python的with语法能帮我们自动关闭资源让代码更整洁。PyMySQL连接对象支持上下文管理协议from contextlib import closing import pymysql with pymysql.connect( host127.0.0.1, userdemo_user, passwordyour_password, databasedemo_db, charsetutf8mb4 ) 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( host127.0.0.1, userdemo_user, passwordyour_password, databasedemo_db, charsetutf8mb4 ) 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占位符千万别用字符串拼接SQLSQL注入事故就是这么来的。参数化查询不光是安全问题它对SQL缓存命中也有好处。4.2 插入数据单条与批量对比插入单条记录import pymysql conn pymysql.connect( host127.0.0.1, userdemo_user, passwordyour_password, databasedemo_db, charsetutf8mb4, autocommitTrue ) cursor conn.cursor() sql INSERT INTO users (username, email, age) VALUES (%s, %s, %s) cursor.execute(sql, (alice, aliceexample.com, 28)) print(插入ID, cursor.lastrowid) conn.close()cursor.lastrowid拿自增ID。如果设置了autocommitFalse需要手动执行conn.commit()。批量插入的效率对比可能是大家最关心的。直接循环单条插入1000条有多慢每条一次网络往返1000次往返延迟全耗在网络上。更好的方式是executemanycursor conn.cursor() data [] for i in range(1000): data.append((fuser_{i}, fuser_{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( host127.0.0.1, userdemo_user, passwordyour_password, databasedemo_db, charsetutf8mb4, cursorclassDictCursor ) 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( host127.0.0.1, userdemo_user, passwordyour_password, databasedemo_db, charsetutf8mb4 ) cursor conn.cursor() # 更新 sql_update UPDATE users SET email %s WHERE username %s cursor.execute(sql_update, (new_emailexample.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引擎是支持事务的我们建表时指定的ENGINEInnoDB就是为事务准备的。5.2 PyMySQL中事务控制的正确姿势PyMySQL的事务控制非常简单import pymysql conn pymysql.connect( host127.0.0.1, userdemo_user, passwordyour_password, databasedemo_db, charsetutf8mb4 ) 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 自动提交设置什么时候开什么时候关很多初学朋友喜欢设置autocommitTrue因为不用手动commit()省心。但实际业务中我劝你谨慎。开启自动提交适合这些场景纯查询逻辑没有写操作非关键数据的写入丢了也没太大关系日志写入、统计上报类的数据必须关闭自动提交的场景涉及多张表的写入操作需要保持数据强一致性的业务订单、支付、库存批量数据导入中间失败要能整体回滚还有一种说法是单条INSERT用不用事务都一样其实不然。InnoDB即使单条INSERT也涉及事务只是自动提交把每条SQL包成了独立事务。关闭自动提交批量插入时最大的收益是减少了写日志redo log的同步次数这也是批量操作性能提升的原因之一。6. 进阶技巧安全、性能与易用性提升6.1 参数化查询彻底告别SQL注入SQL注入是非常危险的漏洞。举例说明# 危险写法千万别这么干 username input(请输入用户名) sql fSELECT * FROM users WHERE username {username} cursor.execute(sql)如果用户输入 OR 11 --拼出来的SQL就变成了SELECT * FROM users WHERE username OR 11 -- 这个查询会把整张表的数据都带出来用户输入被当成了代码执行。正确写法是用占位符sql SELECT * FROM users WHERE username %s cursor.execute(sql, (username,))PyMySQL会把username当纯数据处理永远不参与SQL语法解析。这是底线操作任何动态拼接SQL的行为都必须禁止。6.2 游标类型与遍历效率前面提到DictCursorPyMySQL游标类型其实有几种根据场景选择游标类型返回格式适用场景默认Cursor元组批量数据处理性能优先DictCursor字典业务代码可读性优先SSCursor元组服务端游标超大结果集边取边处理SSDictCursor字典服务端游标超大结果集且需要字典格式SSCursor的机制不一样普通游标是一次性把所有结果拉到客户端SSCursor是在服务端保留游标每取一条拉一条。几十万行数据用普通游标可能直接撑爆内存换SSCursor就能稳定跑完。import pymysql from pymysql.cursors import SSCursor conn pymysql.connect( host127.0.0.1, userdemo_user, passwordyour_password, databasedemo_db, charsetutf8mb4, cursorclassSSCursor ) 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 DBUtilsfrom dbutils.pooled_db import PooledDB import pymysql pool PooledDB( creatorpymysql, maxconnections6, mincached2, maxcached5, blockingTrue, host127.0.0.1, userdemo_user, passwordyour_password, databasedemo_db, charsetutf8mb4 ) # 从连接池拿连接 conn pool.connection() cursor conn.cursor() cursor.execute(SELECT COUNT(*) FROM users) print(cursor.fetchone()) conn.close() # 注意这里是归还连接不是真正关闭maxconnections6表示池子最多持有6个连接高峰期第七个请求会等待blockingTrue直到有连接还回来。这比每次新建连接省下大量握手认证的时间。压测数据在我的机器上连接池方式比每次新建连接吞吐量能提升大约3倍延迟也稳定得多。6.4 常见编码坑emoji和中文乱码MySQL的中文乱码属于经典问题本质是字符集没对齐。三个层面需要一致数据库表的字符集连接字符串的字符集数据的实际编码PyMySQL连接时指定charsetutf8mb4建表时也指定DEFAULT CHARSETutf8mb4基本就能解决。如果还乱码检查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 踩过的坑和对应解决办法整理一张速查表都是实战中容易碰到且搜索频率高的问题报错或现象原因解决方案Cant connect to MySQL server on localhostMySQL服务未启动或IPv6解析问题确认服务状态host改用127.0.0.1Access denied for user xxxlocalhost账号密码错误或权限不足检查密码用GRANT正确授权Unknown database xxx数据库不存在或未创建先创建数据库检查名称拼写Table xxx doesnt exist表不存在或当前连接的数据库不正确检查database参数确认表名Duplicate entry xxx for key uk_username唯一索引冲突先用SELECT判断是否存在或使用INSERT ... ON DUPLICATE KEY UPDATEIncorrect string value: \xF0\x9F\x98\x80字符集不支持emoji表结构、连接参数统一改为utf8mb4Lost connection to MySQL server during query数据包超限或连接空闲太久被断开调整max_allowed_packet或设置wait_timeoutMySQL 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, aliceexample.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, aliceexample.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( hosthost, useruser, passwordpassword, databasedatabase, charsetutf8mb4, cursorclassDictCursor, autocommitFalse ) 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, emailNone, ageNone): 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, limit10): 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, bobexample.com, 25) print(新增用户ID, new_id) user db.get_user(bob) print(查询结果, user) db.update_user(bob, age26) 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这个库本身不算复杂真正的复杂度在如何把连接管理、事务控制、性能优化和业务逻辑组合好。希望这篇文章能帮你少走弯路把更多时间花在业务本身而不是在连接报错里反复挣扎。