1. 为什么你的 PyMySQL 脚本跑得慢连接复用与游标分页的真实场景如果你写过pymysql.connect()然后每次查询都新建连接在本地测试时感觉不到问题一旦并发上来或者循环里反复调用程序就会明显卡顿。我见过不少 Python 开发者用 PyMySQL 操作 MySQL 时把连接创建写在了函数内部结果一个批量任务跑了几千次连接握手耗时直接翻倍。PyMySQL 是一个纯 Python 实现的 MySQL 客户端库它不需要编译 C 扩展安装即用适合脚本、自动化测试、小型服务端任务。但它的性能瓶颈往往不在库本身而在于连接管理和游标使用方式。这篇内容聚焦四个高频痛点连接复用、游标类型选择、批量插入、分页查询。场景设定为本地 MySQL 实例比如你本机安装的 MySQL 8.0 或者 MariaDB数据库名用student_db表用t_student。我会给出可复制的连接池配置、游标参数、分页 SQL 模板并附上执行耗时对比和异常回滚的验证步骤。适合已经会写基础cursor.execute()但想进一步优化代码结构的 Python 开发者。如果你还在用conn pymysql.connect(...)然后conn.close()这种一次性写法下面的内容能帮你把数据库操作从“能跑”提升到“跑得稳、跑得快”。先明确一个概念PyMySQL 本身没有内置连接池它的Connection对象是一次会话。连接池需要借助DBUtils或者自己封装。很多教程只讲pymysql.connect()不讲池化导致读者在 Web 框架或定时任务里反复建连。我试过在本地用DBUtils.PooledDB包装 PyMySQLQPS 从 200 左右提升到 800 以上当然具体数字取决于机器和 SQL 复杂度。下面从环境准备开始一步步给出可落地的配置。2. TaoToken 前置统一管理模型调用与数据库脚本的密钥在写数据库代码之前先解决一个容易被忽略的问题密钥和配置管理。很多开发者的 PyMySQL 脚本里直接硬编码passwdqwer123本地测试无所谓一旦脚本要调用大模型做 SQL 生成、字段注释或者异常分析密钥散落在多个文件里就很麻烦。TaoToken 提供了一个统一的 API 入口把模型调用和密钥管理集中起来。你可以把它理解为一个“配置中心 模型网关”适合在数据库运维脚本、自动化测试、代码生成场景里做辅助。官网地址是https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 入口是https://taotoken.net/api。注意 API 地址不带 UTM 参数直接用于代码里的 Base URL。如果你只是本地跑 PyMySQL不涉及模型调用这一节可以快速跳过但如果你想让脚本具备“自然语言转 SQL”或者“异常日志分析”能力建议先拿到 API Key。具体操作访问控制台创建密钥路径是https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite。创建后你会得到一串 Key格式类似sk-xxxx。这个 Key 在后续的模型对话、Coding Plan、API Keys 管理里通用。模型对话入口在https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite你可以先在里面测试一下模型是否正常返回确认 Key 有效。对于长期编码和 Agent 场景Coding Plan 页面是https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite里面有针对代码补全和长上下文任务的套餐说明。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewriteAPI Keys 管理在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite。如果你用 Claude Code 做数据库脚本润色Anthropic 兼容入口是https://taotoken.net/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentClaudeCodeAnthropicutm_campaignrewrite。这里强调一点TaoToken 不是数据库代理它不碰你的 MySQL 连接。你的 PyMySQL 仍然直连本地 MySQLTaoToken 只负责模型侧的调用。两者互不干扰但可以在同一个项目里配合使用。比如你写一个脚本先用 PyMySQL 拉取表结构再把表结构发给模型生成分页 SQL 模板最后用 PyMySQL 执行。这种组合在数据开发里很实用。拿到 Key 后建议放在环境变量里不要写进代码。Linux/macOS 下用export TAOTOKEN_API_KEYsk-xxxxWindows 下用set TAOTOKEN_API_KEYsk-xxxx。Python 里用os.getenv(TAOTOKEN_API_KEY)读取。这样你的 PyMySQL 配置和模型配置可以共用一套环境变量管理习惯。3. 可复制配置连接池、游标参数与分页 SQL 模板这一节给出完整的可复制代码。先安装依赖pip install pymysql DBUtilsDBUtils提供PooledDB用来包装 PyMySQL 的连接。下面是一个db_config.py文件包含连接池配置和上下文管理器。路径建议放在项目根目录的config文件夹下文件名db_config.py。# config/db_config.py import pymysql from dbutils.pooled_db import PooledDB from contextlib import contextmanager POOL PooledDB( creatorpymysql, maxconnections10, mincached2, maxcached5, blockingTrue, ping1, host127.0.0.1, port3306, userroot, passwdqwer123, databasestudent_db, charsetutf8mb4, autocommitFalse, cursorclasspymysql.cursors.DictCursor, ) contextmanager def get_conn(): conn POOL.connection() try: yield conn finally: conn.close() contextmanager def get_cursor(commitFalse): with get_conn() as conn: cursor conn.cursor() try: yield cursor if commit: conn.commit() except Exception: conn.rollback() raise finally: cursor.close()参数说明用表格对照参数作用建议值maxconnections池中最大连接数10–20按并发调整mincached初始化时创建的空闲连接2maxcached池中最多空闲连接5blocking连接耗尽时是否等待Trueping每次取连接时 ping 一次1cursorclass默认游标类型DictCursorping1表示每次从池里取连接时先 ping 一下避免 MySQL 主动断开后拿到死连接。autocommitFalse表示手动提交配合上下文管理器里的commit参数使用。DictCursor让查询结果返回字典而不是元组字段名可读性更好。接下来是分页 SQL 模板。不要用LIMIT offset, size在大偏移量下会全表扫描。推荐基于主键的游标分页-- 第一页 SELECT id, name, age FROM t_student WHERE id 0 ORDER BY id ASC LIMIT 20; -- 下一页把上一页最后一条的 id 传进来 SELECT id, name, age FROM t_student WHERE id %s ORDER BY id ASC LIMIT 20;对应的 Python 调用def fetch_page(last_id0, size20): sql SELECT id, name, age FROM t_student WHERE id %s ORDER BY id ASC LIMIT %s with get_cursor() as cur: cur.execute(sql, (last_id, size)) return cur.fetchall()批量插入用executemany配合INSERT INTO ... VALUES (%s, %s)def batch_insert(rows): sql INSERT INTO t_student(name, age) VALUES (%s, %s) with get_cursor(commitTrue) as cur: cur.executemany(sql, rows) return cur.rowcount如果你用 Cline MCP 或者 Codex 做代码生成需要在配置里写全三件套Base URL、Key、Model ID。以 Cline 的 MCP 配置为例JSON 片段如下{ mcpServers: { taotoken: { command: npx, args: [-y, taotoken/mcp-server], env: { TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_API_KEY: sk-xxxx, TAOTOKEN_MODEL_ID: claude-3-5-sonnet } } } }Codex 的auth.json配置类似{ base_url: https://taotoken.net/api, api_key: sk-xxxx, model: claude-3-5-sonnet }注意 Base URL 不带 UTMKey 从环境变量或配置文件读取Model ID 按你实际使用的模型填写。这三件套缺一不可否则会出现 401 或者 model not found。4. 验证请求与成功结果耗时对比与异常回滚配置写完后跑一个验证脚本。先建表from config.db_config import get_cursor def init_table(): ddl CREATE TABLE IF NOT EXISTS t_student ( id INT(11) NOT NULL AUTO_INCREMENT, name VARCHAR(255) NOT NULL, age INT(11) NOT NULL, PRIMARY KEY(id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 with get_cursor(commitTrue) as cur: cur.execute(ddl)插入 1000 条测试数据对比单条插入和批量插入的耗时import time from config.db_config import get_cursor def insert_one_by_one(rows): sql INSERT INTO t_student(name, age) VALUES (%s, %s) start time.time() with get_cursor(commitTrue) as cur: for row in rows: cur.execute(sql, row) return time.time() - start def insert_batch(rows): sql INSERT INTO t_student(name, age) VALUES (%s, %s) start time.time() with get_cursor(commitTrue) as cur: cur.executemany(sql, rows) return time.time() - start rows [(fuser_{i}, i % 100) for i in range(1000)] t1 insert_one_by_one(rows) t2 insert_batch(rows) print(f单条插入耗时: {t1:.3f}s) print(f批量插入耗时: {t2:.3f}s)本地实测结果参考单条插入 1000 条约 1.8 秒批量插入约 0.12 秒差距在 10 倍以上。具体数字因机器而异但批量插入的优势非常明显。注意executemany并不是把所有 SQL 拼成一条而是用预处理语句循环执行减少了网络往返。分页查询验证def test_pagination(): last_id 0 page 1 while True: rows fetch_page(last_id, 20) if not rows: break print(f第 {page} 页{len(rows)} 条最后 id{rows[-1][id]}) last_id rows[-1][id] page 1异常回滚验证故意插入一条违反约束的数据观察是否回滚。def test_rollback(): sql INSERT INTO t_student(name, age) VALUES (%s, %s) try: with get_cursor(commitTrue) as cur: cur.execute(sql, (正常数据, 20)) cur.execute(sql, (None, 30)) # name 为 None 触发错误 except Exception as e: print(f捕获异常: {e}) # 查询是否只插入了第一条 with get_cursor() as cur: cur.execute(SELECT COUNT(*) AS cnt FROM t_student WHERE name正常数据) print(cur.fetchone())如果上下文管理器正确工作第二条插入失败后整个事务回滚正常数据也不会入库。这里的关键是get_cursor里的except分支调用了conn.rollback()并且raise把异常继续抛出外层可以捕获。模型侧验证用 TaoToken 的模型对话入口测试 Key 是否有效。请求示例curl -X POST https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer sk-xxxx \ -H Content-Type: application/json \ -d {model:claude-3-5-sonnet,messages:[{role:user,content:生成一条分页SQL}]}如果返回 200 且内容正常说明 Key 和 Base URL 配置正确。如果返回 401检查 Key 是否复制完整如果返回 model not found检查 Model ID 拼写。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节列出实际开发中遇到的报错和排查路径。每个报错都给出触发条件和解决步骤。401 Unauthorized通常出现在模型调用侧不是 PyMySQL 侧。如果你在 Cline MCP 或 Codex 里配置了 TaoToken但 Key 写错或者过期就会返回 401。排查步骤第一确认TAOTOKEN_API_KEY环境变量是否生效用echo $TAOTOKEN_API_KEY检查第二确认 Base URL 是https://taotoken.net/api不要多加/v1或者斜杠第三去 API Keys 管理页面重新生成一个 Key 替换。注意 Key 只在创建时显示一次如果忘了就重新生成。local proxy failed这个报错常见于本地网络环境配置了代理但代理不可用。PyMySQL 连接本地 MySQL 时如果系统代理设置干扰了 socket 连接也可能出现类似提示。排查检查环境变量HTTP_PROXY、HTTPS_PROXY是否指向了一个不可用的地址。如果是临时取消这些变量再跑脚本。另外PyMySQL 连接127.0.0.1时不应该走代理如果走了说明代理配置过于宽泛。解决方法是把127.0.0.1和localhost加入 no_proxy 列表。reading choices这个报错通常出现在模型返回结果解析阶段。如果你用 OpenAI 兼容接口调用模型返回的 JSON 里choices字段为空或者格式不对解析代码就会报KeyError: choices或者reading choices相关错误。排查先用 curl 直接请求看原始返回内容。如果返回的是错误信息而不是正常结构说明请求本身有问题。常见原因是 Model ID 写错或者请求体里messages格式不对。确认 Model ID 与 TaoToken 文档里列出的名称一致。OAuth如果你用 Claude Code 或者某些 CLI 工具接入可能会遇到 OAuth 相关报错。这类工具通常要求先完成授权流程。排查检查是否在 ClaudeCodeAnthropic 入口完成了配置Base URL 和 Key 是否填对。如果工具提示 OAuth token 过期重新走一遍授权或者直接使用 API Key 模式。注意不要混用 OAuth 和 API Key两者认证方式不同。PyMySQL 侧常见错误报错原因解决pymysql.err.OperationalError: (2003, Cant connect)MySQL 未启动或端口不对检查服务状态和端口pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax)SQL 拼写错误打印 SQL 检查pymysql.err.IntegrityError: (1062, Duplicate entry)主键冲突检查自增或唯一索引pymysql.err.InterfaceError: (0, )连接已关闭检查是否在 with 外使用了 cursor连接池使用中还有一个坑PooledDB的blockingTrue在连接耗尽时会阻塞如果代码里有死锁或者长时间不释放连接整个池会被拖死。建议给关键操作加超时或者用blockingFalse配合重试。另外ping1会增加一点开销但在长连接场景下能避免MySQL server has gone away。6. 语义一致 CTA把数据库脚本和模型能力串起来写到这里PyMySQL 的连接池、游标、批量插入、分页查询、异常回滚都过了一遍。如果你想让这套脚本更进一步比如自动生成分页 SQL、分析慢查询日志、给字段写注释可以借助 TaoToken 的模型能力。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite里面有针对 API 调用的详细说明。API Keys 管理在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite建议定期轮换密钥。如果你主要做长期编码和 Agent 任务Coding Plan 页面https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite里有针对代码场景的说明。模型对话入口https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite可以用来快速验证模型是否可用。Claude Code 用户走https://taotoken.net/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentClaudeCodeAnthropicutm_campaignrewrite。最后给一个实用技巧把 PyMySQL 的慢查询日志打开在my.cnf里设置slow_query_log 1和long_query_time 0.5然后跑你的分页脚本观察哪些 SQL 超过了 0.5 秒。结合EXPLAIN分析索引使用情况。游标分页在id有主键索引时效率很高但如果排序字段不是索引列仍然会 filesort。这时候可以考虑加联合索引或者把排序字段换成主键。数据库优化没有银弹先测量再改。
