1. 从一次真实的数据落库失败说起PyMySQL 连接与事务的坑很多 Python 开发者第一次用 PyMySQL 读写 MySQL 时代码看起来完全正确跑起来却报pymysql.err.OperationalError或者数据「查不到」。我试过在一个数据采集脚本里连续插入 2000 条记录程序没报错但去数据库一查只有 0 条——原因就是忘了commit()。PyMySQL 默认关闭了自动提交所有写操作都挂在一个未提交的事务里连接一关数据全回滚。PyMySQL 是什么它是一个纯 Python 实现的 MySQL 客户端库不需要编译 C 扩展pip install pymysql就能用兼容 MySQL 5.7/8.0 和 MariaDB。它能做什么连接数据库、执行 SQL、参数化查询、批量写入、事务控制、读取结果集基本覆盖日常 CRUD 全部需求。适合谁适合做数据同步脚本、后端服务、爬虫落库、报表导出这类需要直接操作 MySQL 的 Python 场景。但实际项目里除了数据库连接本身还有一个容易被忽略的环节调用外部模型或 API 时的凭证管理。比如你在数据清洗流程里接一个大模型做文本分类每个脚本里硬编码一堆 Key换环境就崩。这篇会把 PyMySQL 的完整读写链路讲透同时演示怎么用 TaoToken 的统一 Key 通道管理调用凭证让数据库操作和 API 调用两件事都不再散落各处。核心检索词先明确PyMySQL 读写 MySQL、参数化查询、批量写入、事务提交、统一 Key 接入。下面从连接配置开始一步步给出可复制的代码。2. TaoToken 前置准备统一 Key 与 API 通道配置在写 PyMySQL 代码之前先把凭证管理这件事理清楚。TaoToken 是一个统一 Key/API 通道管理平台你可以把它理解成一个「凭证中转站」把不同来源的模型调用凭证集中托管代码里只引用一个统一 Key换环境、换模型时不用改业务代码。官网入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 地址https://taotoken.net/api具体操作路径先注册登录进入控制台创建 API Key然后在「模型对话」页面可以测试模型是否可用。如果你是长期做编码或 Agent 开发可以看 Coding Plan 页面需要管理多个 Key 就去 API Keys 页面接入细节看文档页。拿到 Key 之后建议不要硬编码在脚本里用环境变量或.env文件管理。下面是一个.env示例# .env TAOTOKEN_API_KEYsk-你的统一Key TAOTOKEN_BASE_URLhttps://taotoken.net/api MYSQL_HOST127.0.0.1 MYSQL_PORT3306 MYSQL_USERroot MYSQL_PASSWORD123456 MYSQL_DBtest然后在 Python 里用python-dotenv加载import os from dotenv import load_dotenv load_dotenv() TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY) TAOTOKEN_BASE_URL os.getenv(TAOTOKEN_BASE_URL)这样数据库连接信息和 API 凭证都从环境读取本地、测试、生产三套环境只需要换.env代码零改动。这一步做完后面 PyMySQL 的连接配置就可以直接复用这些变量。注意API Key 属于敏感凭证不要提交到 Git 仓库.env要写进.gitignore。3. 可复制配置PyMySQL 连接串与 CRUD 完整代码这一节给出可以直接复制运行的完整代码。先安装依赖pip install pymysql python-dotenv3.1 连接配置与建表import pymysql import os from dotenv import load_dotenv load_dotenv() def get_conn(): return pymysql.Connect( hostos.getenv(MYSQL_HOST, 127.0.0.1), portint(os.getenv(MYSQL_PORT, 3306)), useros.getenv(MYSQL_USER, root), passwdos.getenv(MYSQL_PASSWORD, ), dbos.getenv(MYSQL_DB, test), charsetutf8mb4, cursorclasspymysql.cursors.DictCursor, ) def init_table(): conn get_conn() cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS student ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(64) NOT NULL, age INT, register_date DATE, gender CHAR(1) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; ) conn.commit() cursor.close() conn.close()这里有几个关键参数charsetutf8mb4保证中文和 emoji 不乱码cursorclassDictCursor让查询结果直接返回字典比元组好读ENGINEInnoDB支持事务这是后面 commit 生效的前提。3.2 参数化查询与批量写入def insert_one(name, age, register_date, gender): conn get_conn() cursor conn.cursor() sql INSERT INTO student(name, age, register_date, gender) VALUES (%s, %s, %s, %s) rows cursor.execute(sql, (name, age, register_date, gender)) conn.commit() cursor.close() conn.close() return rows def insert_many(data): conn get_conn() cursor conn.cursor() sql INSERT INTO student(name, age, register_date, gender) VALUES (%s, %s, %s, %s) rows cursor.executemany(sql, data) conn.commit() cursor.close() conn.close() return rows参数化查询用%s占位不要把变量拼进 SQL 字符串否则会有注入风险。executemany适合批量场景比循环单条插入快很多。3.3 事务控制与异常回滚def transfer_insert(data): conn get_conn() cursor conn.cursor() try: sql INSERT INTO student(name, age, register_date, gender) VALUES (%s, %s, %s, %s) cursor.executemany(sql, data) conn.commit() except Exception as e: conn.rollback() print(写入失败已回滚:, e) raise finally: cursor.close() conn.close()事务三件套commit()提交、rollback()回滚、finally里关闭资源。只要有一批数据里任意一条失败整批回滚保证数据一致性。3.4 统一 Key 调用示例如果你在数据流程里还要调用模型可以用统一 Key 走 OpenAI 兼容接口from openai import OpenAI client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), ) resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 把这条学生记录转成 JSON}], ) print(resp.choices[0].message.content)Base URL、Key、Model ID 三件套齐全换模型只改model字段凭证不用动。4. 验证请求一条查询与一条写入确认数据落库配置写完必须验证。先跑一条写入再跑一条查询看数据是否真的进库。if __name__ __main__: init_table() # 单条写入 affected insert_one(xiaoming, 23, 2018-12-30, M) print(单条写入受影响行数:, affected) # 批量写入 data [ (student1, 23, 2018-01-31, M), (student2, 24, 2018-02-27, M), (student3, 28, 2018-03-31, F), (student4, 26, 2018-04-30, M), ] affected insert_many(data) print(批量写入受影响行数:, affected) # 查询验证 conn get_conn() cursor conn.cursor() cursor.execute(SELECT * FROM student ORDER BY id DESC LIMIT 5) for row in cursor.fetchall(): print(row) cursor.close() conn.close()预期输出类似单条写入受影响行数: 1 批量写入受影响行数: 4 {id: 5, name: student4, age: 26, register_date: datetime.date(2018, 4, 30), gender: M} {id: 4, name: student3, age: 28, register_date: datetime.date(2018, 3, 31), gender: F} ...看到affected大于 0 且查询能返回刚插入的记录说明写入和事务提交都正常。如果查询为空先检查是否漏了commit()再检查连接是否指向了正确的库。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth实际跑的时候报错集中在几个地方逐个对照。报错一pymysql.err.OperationalError: (1045, Access denied for user)这是数据库账号密码错误不是 API Key 问题。检查.env里的MYSQL_USER和MYSQL_PASSWORD确认该用户对目标库有权限。报错二401 Unauthorized或invalid api key调用模型接口时出现说明统一 Key 无效或没加载。检查TAOTOKEN_API_KEY是否从环境变量正确读取Key 是否过期。可以到 API Keys 页面重新生成。报错三local proxy failed或连接超时通常是网络层问题检查TAOTOKEN_BASE_URL是否写成https://taotoken.net/api不要多加路径或斜杠。数据库侧则检查MYSQL_HOST和端口是否可达。报错四reading choices相关报错如KeyError: choices说明接口返回结构不符合预期多半是model字段写错或该模型不支持当前调用方式。对照文档页确认 Model ID再重试。报错五OAuth 或鉴权流程报错如果你用的是需要 OAuth 的客户端如 Claude Code 类工具确认凭证类型和接口匹配。统一 Key 走的是 API Key 模式不要混用 OAuth 流程。报错六数据查不到但没报错九成是漏了commit()。PyMySQL 默认不自动提交写操作必须显式提交。另一个可能是连接了错误的库检查MYSQL_DB。排查顺序建议先确认数据库连接通不通再确认 API Key 有没有加载最后看返回结构。把每一步的异常打印出来定位会快很多。6. 把凭证和数据库操作都收进一套流程PyMySQL 的读写链路本身不复杂连接、游标、参数化 SQL、提交、关闭。真正容易出问题的是凭证散落和环境切换。把数据库连接信息和统一 Key 都放进.env代码里只引用变量本地和生产就能用同一套逻辑。如果你还在手动管理多个 Key建议去控制台把凭证集中起来再配合接入文档把调用方式固定下来。需要验证模型是否可用直接去模型对话页面测一条长期做编码或 Agent 开发可以看 Coding Plan 页面规划用量。数据库这边记住三件事参数化查询防注入、批量写入用executemany、写操作必须commit()。把这三条守住PyMySQL 的坑基本就绕过去了。