pymysql单条插入数据和批量插入数据：用TaoToken统一Key跑通两种写入路径

1. pymysql 单条插入与批量插入到底差在哪一次本地 MySQL 写入路径实测pymysql 单条插入数据和批量插入数据是 Python 操作 MySQL 时最常被拿来对比的两条写入路径。单条插入就是一行一条INSERT写起来直观、报错定位快批量插入是把多行数据打包成一次executemany让驱动和数据库在一次网络往返里处理多行。适合谁适合正在做数据采集落库、日志归档、订单明细导入或者单纯想搞清楚「为什么我循环 insert 一万行要等半分钟」的开发者。我这次在本地 MySQL 环境里用同一张表、同一批数据分别跑单条插入和批量插入记录耗时、校验行数并且把调用凭证统一交给 TaoToken 的 Key/API 通道管理。为什么要扯到凭证管理因为真实项目里数据库连接参数、模型调用参数经常散落在各个脚本里改一次要翻好几个文件。把凭证收敛到一处脚本只读环境变量迁移和轮换都省事。先说结论方向数据量小的时候两者差别不明显一旦到几千行以上批量插入的优势会非常直观差距往往在一个数量级。但批量插入不是无脑用executemany的参数绑定写法、批次大小、事务提交时机都会影响结果。下面从环境准备开始一步步把两种写法都跑通再对照耗时和行数验证效果。这篇会给出可直接复制的 Python 脚本、executemany参数绑定示例、耗时对比代码以及常见报错排查。你跟着做能在自己的机器上复现出同样的对比结果。2. 用 TaoToken 统一管理调用凭证本地 MySQL 写入前的准备在写插入脚本之前先把凭证这件事理顺。很多人的脚本里直接写死pymysql.connect(localhost,root,123,testdb)本地玩玩没问题一旦脚本变多、环境变多密码散落各处就是隐患。我的做法是把数据库连接信息和外部 API 凭证都收敛到环境变量脚本里只引用变量名。TaoToken 在这里扮演的是统一 Key/API 通道的角色。你可以把它理解成一个凭证中转站本地脚本、编码工具、Agent 需要调用外部能力时都走同一个 Base URL 和同一把 Key不用每个工具单独配一套。官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 这个不加 UTM。具体到操作先去控制台创建 API Key。控制台地址带 deep linkhttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。创建完 Key 之后在 API Keys 页面可以查看和管理https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。如果你后面要接 Claude Code 这类编码工具文档页在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite Claude Code 的接入说明在 https://taotoken.net/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentClaudeCodeAnthropicutm_campaignrewrite 。把 Key 写进环境变量Linux/macOS 下可以这样export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export MYSQL_HOST127.0.0.1 export MYSQL_PORT3306 export MYSQL_USERroot export MYSQL_PASSWORD你的密码 export MYSQL_DBtestdbWindows PowerShell 用$env:TAOTOKEN_API_KEYsk-你的key这种写法。这样脚本里os.environ.get(MYSQL_PASSWORD)就能拿到不用硬编码。注意环境变量只在当前终端会话有效想持久化就写进~/.bashrc或系统环境变量设置里。别把 Key 提交到 Git 仓库。数据库这边先建一张测试表。字段类型要和插入数据对应这是后面参数绑定不出错的前提CREATE TABLE IF NOT EXISTS employee ( id INT AUTO_INCREMENT PRIMARY KEY, first_name VARCHAR(50), last_name VARCHAR(50), age INT, sex CHAR(1), income DECIMAL(10,2), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;装依赖就一条命令pip install pymysql。确认版本pip show pymysql1.0 以上都行。准备工作到这里就够了接下来进入代码环节。3. 可复制配置单条插入与 executemany 批量插入脚本这一节给两段完整脚本都能直接跑。先看单条插入。核心是cursor.execute(sql, params)注意参数用占位符%s不要把值拼进 SQL 字符串——拼接既有注入风险又容易因为引号、类型不对报错。import os import time import pymysql conn pymysql.connect( hostos.environ.get(MYSQL_HOST, 127.0.0.1), portint(os.environ.get(MYSQL_PORT, 3306)), useros.environ.get(MYSQL_USER, root), passwordos.environ.get(MYSQL_PASSWORD, ), databaseos.environ.get(MYSQL_DB, testdb), charsetutf8mb4, ) rows [ (Mac, Mohan, 20, M, 2000), (Li, Si, 16, F, 1000), (Bruse, Jerry, 30, F, 3000), (Lee, Tomcat, 40, M, 4000), (Zhang, San, 18, M, 1500), ] sql INSERT INTO employee (first_name, last_name, age, sex, income) VALUES (%s, %s, %s, %s, %s) start time.perf_counter() try: with conn.cursor() as cursor: for row in rows: cursor.execute(sql, row) conn.commit() except Exception as e: conn.rollback() print(单条插入失败:, e) finally: conn.close() print(f单条插入 {len(rows)} 行耗时 {time.perf_counter() - start:.4f} 秒)再看批量插入。关键区别是cursor.executemany(sql, val)val是一个可迭代对象每个元素是一行参数元组。SQL 里同样用%s占位不要加引号import os import time import pymysql conn pymysql.connect( hostos.environ.get(MYSQL_HOST, 127.0.0.1), portint(os.environ.get(MYSQL_PORT, 3306)), useros.environ.get(MYSQL_USER, root), passwordos.environ.get(MYSQL_PASSWORD, ), databaseos.environ.get(MYSQL_DB, testdb), charsetutf8mb4, ) val ( (li, si, 16, F, 1000), (Bruse, Jerry, 30, F, 3000), (Lee, Tomcat, 40, M, 4000), (zhang, san, 18, M, 1500), ) sql INSERT INTO employee (first_name, last_name, age, sex, income) VALUES (%s, %s, %s, %s, %s) start time.perf_counter() try: with conn.cursor() as cursor: cursor.executemany(sql, val) conn.commit() except Exception as e: conn.rollback() print(批量插入失败:, e) finally: conn.close() print(f批量插入 {len(val)} 行耗时 {time.perf_counter() - start:.4f} 秒)如果你要把 TaoToken 的凭证也纳入统一配置可以写一个settings.json放在项目根目录脚本启动时读取。这样数据库和 API 凭证都在一处{ taotoken: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY }, mysql: { host: 127.0.0.1, port: 3306, user: root, password_env: MYSQL_PASSWORD, database: testdb, charset: utf8mb4 } }脚本里用json.load读进来密码字段只存环境变量名不存明文。这套结构在 Cline MCP、Codex 的auth.json场景里也通用Base URL、Key、Model ID 三件套固定下来换工具只改一处。Codex 的auth.json通常长这样把base_url指向统一通道即可{ base_url: https://taotoken.net/api, api_key: sk-你的key, model: 你的模型ID }注意executemany的 SQL 里写VALUES (%s, %s, %s, %s, %s)是错的引号会让%s变成普通字符串插入的是字面量%s。正确写法是全部裸%s。4. 验证请求与成功结果耗时对比和行数校验代码跑通只是第一步得用数据证明批量插入确实更快。我写了一个对比脚本生成 5000 行测试数据分别用单条和批量插入记录耗时最后查表校验行数。import os import time import pymysql def get_conn(): return pymysql.connect( hostos.environ.get(MYSQL_HOST, 127.0.0.1), portint(os.environ.get(MYSQL_PORT, 3306)), useros.environ.get(MYSQL_USER, root), passwordos.environ.get(MYSQL_PASSWORD, ), databaseos.environ.get(MYSQL_DB, testdb), charsetutf8mb4, ) def gen_rows(n): return [(ffirst{i}, flast{i}, 20 i % 40, M if i % 2 else F, 1000 i) for i in range(n)] SQL INSERT INTO employee (first_name, last_name, age, sex, income) VALUES (%s, %s, %s, %s, %s) def insert_one_by_one(rows): conn get_conn() start time.perf_counter() try: with conn.cursor() as cur: for r in rows: cur.execute(SQL, r) conn.commit() finally: conn.close() return time.perf_counter() - start def insert_batch(rows, batch_size1000): conn get_conn() start time.perf_counter() try: with conn.cursor() as cur: for i in range(0, len(rows), batch_size): cur.executemany(SQL, rows[i:i batch_size]) conn.commit() finally: conn.close() return time.perf_counter() - start if __name__ __main__: n 5000 rows gen_rows(n) t1 insert_one_by_one(rows) print(f单条插入 {n} 行耗时: {t1:.3f} 秒) t2 insert_batch(rows) print(f批量插入 {n} 行耗时: {t2:.3f} 秒) print(f批量比单条快 {t1 / t2:.1f} 倍) conn get_conn() with conn.cursor() as cur: cur.execute(SELECT COUNT(*) FROM employee) print(表内总行数:, cur.fetchone()[0]) conn.close()实测下来5000 行数据单条插入通常在几秒到十几秒批量插入往往不到一秒差距明显。行数校验那一步很重要如果executemany中途异常被吞掉行数会对不上。跑完看到总行数等于两次插入之和才算真正成功。批次大小也值得调。batch_size1000是个稳妥起点太小退化成接近单条太大可能撞上max_allowed_packet。你可以把batch_size改成 100、500、2000 各跑一遍看耗时曲线。一般到某个点之后收益就平了。注意对比测试前先TRUNCATE TABLE employee清空否则行数校验会叠加历史数据结论失真。5. 本篇常见错排查401、local proxy failed、reading choices 与 OAuth写入脚本本身报错和凭证通道报错是两类问题分开看。数据库侧最常见的几个pymysql.err.OperationalError: (1045, Access denied for user ...)这是账号密码不对。检查环境变量MYSQL_PASSWORD是否真的注入到当前进程print(os.environ.get(MYSQL_PASSWORD))打一下。注意密码里有特殊字符时shell 里要加引号。pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax)多半是executemany的 SQL 里带了引号或者列数和参数个数对不上。数一下VALUES里的%s个数和每行元组长度是否一致。pymysql.err.DataError: (1406, Data too long for column)字段长度不够比如sex CHAR(1)塞了Female。改表结构或截断数据。凭证通道侧如果你在接编码工具或 Agent 时遇到这些401 UnauthorizedKey 无效或没带上。确认请求头里Authorization: Bearer sk-xxx格式正确Key 没有多余空格。去 API Keys 页面核对 Key 状态https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。local proxy failed通常是本地网络配置或代理设置干扰了请求。检查系统代理、环境变量HTTP_PROXY/HTTPS_PROXY是否指向了不可用的地址清掉再试。reading choices这类报错一般出现在解析模型返回结构时返回体不是预期的 JSON 结构。先打印原始响应体看内容确认 Base URL 拼对了——https://taotoken.net/api后面接的路径要和文档一致别自己拼错。OAuth相关报错多出现在 Claude Code 这类工具的登录流程里。如果你用的是 Key 方式接入就不该走 OAuth 分支检查工具配置里是不是选错了认证模式。Claude Code 的接入细节看 https://taotoken.net/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentClaudeCodeAnthropicutm_campaignrewrite 。排查顺序建议先确认数据库能连上mysql -u root -p手动登一次再确认脚本参数绑定对最后才怀疑凭证通道。分层定位别一上来就改 Key。6. 把两种写入路径固定成模板后续怎么复用单条插入和批量插入不是二选一而是按场景选。数据量小、需要逐行拿到插入结果或错误行号时单条更合适数据量大、追求吞吐时批量插入是默认选择。我现在的习惯是写一个insert_rows(rows, batch_sizeNone)函数batch_size为None就走单条传数字就走executemany分批调用方按数据量决定。凭证这块把 TaoToken 的 Base URL、Key、Model ID 三件套和数据库连接信息都收进settings.json加环境变量脚本里不出现明文。这样换机器、换环境、接新工具时只改配置文件一处。模型对话类调试可以用 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 快速验证通道是否通长期跑编码或 Agent 任务Coding Plan 页面在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。最后留一个实用技巧批量插入时给executemany外面套一层计时和行数断言跑完立刻SELECT COUNT(*)对账。数据导入这种活宁可多花两秒校验也别等下游发现少了几百行再回头查。把校验写进脚本比事后排查省心得多。