1. 批量插入卡在哪儿从 execute 循环到 executemany如果你写过 Python 操作 MySQL大概率经历过这个阶段拿到一个列表里面几百上千条数据然后写个 for 循环一条一条cursor.execute()插进去。数据量小的时候没感觉一旦上万条程序跑起来就像老牛拉车几分钟都跑不完。pymysql提供的executemany方法就是来解决这个问题的。它允许你把 SQL 语句的占位符写一次然后把所有数据打包成一个列表传进去驱动层会帮你批量执行。相比循环单条插入网络往返次数大幅减少速度提升非常明显。但实际用起来坑也不少。比如占位符写错、参数结构不对、事务没提交、字符集乱码还有连接配置散落在代码各处换个环境就要改一堆地方。这篇就围绕pymysql的executemany批量插入把可复制的配置骨架、参数绑定示例、验证动作和报错排查一次讲清楚。适合正在写数据导入脚本、爬虫入库、日志批量落库的 Python 开发者。2. 用 TaoToken 统一管理 Key 与 API 通道在讲数据库操作之前先解决一个工程化问题配置管理。很多人的脚本里数据库密码、API Key、模型地址都是硬编码的代码一提交就泄露换台机器就要手动改。我试过把这类敏感配置抽到一个settings.json里代码只读配置不碰明文。TaoToken 在这里的角色是统一 Key 和 API 通道的管理入口。你可以把它理解成一个配置中枢数据库连接参数、模型调用的 API Key、不同环境的地址都收敛到一份配置文件里。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基础地址是 https://taotoken.net/api 。需要先拿到 Key 的话去控制台创建https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 然后在 API Keys 页面生成https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 遇到参数问题可以对照查。注意数据库密码和 API Key 都属于敏感信息settings.json不要提交到公开仓库建议加入.gitignore或者用环境变量覆盖。3. settings.json 骨架与 pymysql 连接封装先给一份可以直接抄的settings.json骨架。结构上分三块数据库连接、TaoToken 通道、批量插入参数。字段名你可以按自己习惯改但建议保持层级清晰。{ database: { host: 127.0.0.1, port: 3306, user: root, password: your_password, database: pytest3, charset: utf8mb4 }, taotoken: { api_base: https://taotoken.net/api, api_key: sk-你的Key, model: claude-sonnet-4-5 }, batch: { chunk_size: 500, table: aaa } }这里charset用utf8mb4而不是utf8因为 MySQL 的utf8实际是三个字节存不了 emoji 和部分生僻字utf8mb4才是真正的四字节 UTF-8。这个坑我在存用户昵称时踩过插入报错或者变成问号换成utf8mb4就好了。接着写一个读取配置并建立连接的封装import json import pymysql def load_settings(pathsettings.json): with open(path, r, encodingutf-8) as f: return json.load(f) def get_conn(settings): db settings[database] return pymysql.connect( hostdb[host], portdb[port], userdb[user], passworddb[password], databasedb[database], charsetdb[charset], cursorclasspymysql.cursors.DictCursor )cursorclass设成DictCursor后查询结果会以字典返回字段名直接当 key 用调试时比元组直观很多。批量插入本身不依赖这个但排查问题时方便。4. executemany 参数绑定与批量插入示例executemany的签名是cursor.executemany(sql, args)。第一个参数是带占位符的 SQL第二个参数是一个可迭代对象每个元素对应一条记录。关键点在于占位符只写%s不要加引号。很多人从execute迁移过来时习惯写成values(%s, %s)给字符串占位符套了引号。在executemany里这会出问题因为驱动会把引号当成字面量导致插入的值变成带引号的字符串或者直接报参数数量不匹配。正确的写法def batch_insert(conn, table, rows, chunk_size500): sql fINSERT INTO {table}(id, name) VALUES(%s, %s) cursor conn.cursor() total 0 for i in range(0, len(rows), chunk_size): chunk rows[i:i chunk_size] affected cursor.executemany(sql, chunk) total affected conn.commit() cursor.close() return total if __name__ __main__: settings load_settings() conn get_conn(settings) data [(i, f学生{i}) for i in range(20, 30)] n batch_insert(conn, settings[batch][table], data) print(f插入完成影响行数{n}) conn.close()这里做了分块处理。为什么不一次性把十万条全塞进去因为executemany底层会把所有参数拼成一条大 SQL 或者分批发送数据量太大时可能撞上max_allowed_packet限制报Packet too large。分块到 500 或 1000 条一批既快又稳。参数结构上rows是列表每个元素是元组元组里的值按位置对应 SQL 里的%s。顺序不能错(id, name)对应VALUES(%s, %s)第一个%s拿 id第二个拿 name。5. 验证批量插入是否成功插入完别急着关连接先验证一下。两种方式一是看executemany的返回值它返回受影响的行数二是直接查库确认。def verify(conn, table, start_id, end_id): cursor conn.cursor() cursor.execute( fSELECT id, name FROM {table} WHERE id BETWEEN %s AND %s ORDER BY id, (start_id, end_id) ) rows cursor.fetchall() for r in rows: print(r) cursor.close() return rows跑一遍完整流程预期输出类似插入完成影响行数10 {id: 20, name: 学生20} {id: 21, name: 学生21} ... {id: 29, name: 学生29}如果影响行数是 10查询也能查到 20 到 29 这十条说明批量插入成功。如果影响行数是 0但没报错大概率是事务没提交或者插入的数据和现有主键冲突被忽略了。提示executemany默认不会自动提交必须显式调用conn.commit()。用with conn.cursor() as cursor也不会自动提交事务控制要自己管。6. 常见报错定位与排查步骤批量插入报错时别慌按下面几个方向逐个排查。报错一(1064, You have an error in your SQL syntax)先检查 SQL 里的占位符。executemany只认%s写成?、:name或者%d都会语法错误。另外表名、字段名如果和 MySQL 关键字冲突要用反引号包起来比如order。报错二(1136, Column count doesnt match value count)SQL 里写了两个%s但元组里给了三个值或者反过来。数一下VALUES后面的占位符个数和每个元组的长度对齐。用DictCursor时不影响插入但参数还是按位置匹配。报错三(1406, Data too long for column)字段长度不够或者字符集不对。比如字段是varchar(10)你插了 15 个字符。检查建表语句的字段长度以及charset是否设成了utf8mb4。报错四(2006, MySQL server has gone away)或Packet too large连接超时或单次数据包过大。前者调大wait_timeout后者减小chunk_size或者调大 MySQL 的max_allowed_packet。分块插入是最省事的办法。报错五插入成功但数据是乱码连接时charset没设对或者表本身的字符集不是utf8mb4。建表时指定DEFAULT CHARSETutf8mb4连接参数也保持一致。排查时建议打开pymysql的调试日志或者在executemany外面包一层 try-except把chunk的前几条打印出来看看参数结构对不对。try: cursor.executemany(sql, chunk) except Exception as e: print(出错批次首条数据, chunk[0]) print(错误信息, e) raise7. 配置与接入入口数据库连接和批量插入调通之后如果你还想把模型调用也统一到同一套配置里比如用脚本自动生成测试数据、或者让模型帮忙分析插入日志可以走 TaoToken 的通道。模型对话入口在 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 长期跑编码任务或 Agent 场景可以看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入文档里对参数和返回结构写得比较细遇到 Key 鉴权、模型名不对、请求超时这类问题对照文档排查比瞎试快得多https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。API Key 统一在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 管理换环境时只改settings.json里的api_key字段就行代码不用动。最后留一个实用习惯批量插入脚本跑生产之前先在测试库跑一遍把chunk_size调到 100 左右观察耗时和内存占用再逐步放大。我一般从 500 起步超过 2000 就分块稳当。