Python 批量插入多条数据：pymysql executemany 方法配 TaoToken 的 settings.json 骨架与报错排查-平芜编程栈

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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 基础地址是 https://taotoken.net/api 。

需要先拿到 Key 的话，去控制台创建：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，然后在 API Keys 页面生成：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，遇到参数问题可以对照查。

注意：数据库密码和 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(path="settings.json"): with open(path, "r", encoding="utf-8") as f: return json.load(f) def get_conn(settings): db = settings["database"] return pymysql.connect( host=db["host"], port=db["port"], user=db["user"], password=db["password"], database=db["database"], charset=db["charset"], cursorclass=pymysql.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_size=500): sql = f"INSERT 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( f"SELECT 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 doesn't 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 CHARSET=utf8mb4，连接参数也保持一致。

排查时建议打开pymysql的调试日志，或者在executemany外面包一层 try-except，把chunk的前几条打印出来，看看参数结构对不对。

try: cursor.executemany(sql, chunk) except Exception as e: print("出错批次首条数据：", chunk[0]) print("错误信息：", e) raise

7. 配置与接入入口

数据库连接和批量插入调通之后，如果你还想把模型调用也统一到同一套配置里，比如用脚本自动生成测试数据、或者让模型帮忙分析插入日志，可以走 TaoToken 的通道。模型对话入口在 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite ，长期跑编码任务或 Agent 场景可以看 Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。

接入文档里对参数和返回结构写得比较细，遇到 Key 鉴权、模型名不对、请求超时这类问题，对照文档排查比瞎试快得多：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。API Key 统一在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 管理，换环境时只改settings.json里的api_key字段就行，代码不用动。

最后留一个实用习惯：批量插入脚本跑生产之前，先在测试库跑一遍，把chunk_size调到 100 左右，观察耗时和内存占用，再逐步放大。我一般从 500 起步，超过 2000 就分块，稳当。