1. 本地脚本连 MySQL 总踩坑先看清 pymysql 逐行注释到底解决什么问题如果你写过 Python 操作 MySQL 的脚本大概率遇到过这几种情况连接参数写错一个字母报Access deniedSQL 执行完没commit数据查不到游标用完没关跑几百次之后连接池爆掉。这些问题单独看都不难但混在一个几十行的脚本里排查起来就很烦。pymysql 逐行注释的写法核心目的就是让每一行代码的意图都摆在明面上——连接参数为什么这么填、游标是干什么的、commit 和 rollback 分别在什么时机用、fetchone 和 fetchall 的区别在哪。你把这些注释写清楚下次改脚本或者交给同事维护时不用再从头猜逻辑。这篇内容面向的是本地脚本和小型服务场景比如定时跑的数据同步任务、内部工具的后台查询、轻量级 API 的数据库层。不涉及分布式事务、连接池调优这些重话题重点放在“能跑起来、能看懂、能改对”上。我会先给出一套完整的连接配置和建表语句然后逐行注释地写增删改查接着说明怎么把脚本里的 API 调用统一走 TaoToken 的 Key 通道最后用一条查询验证连通和返回结果。你跟着敲一遍基本就能把 pymysql 的常用操作串起来。先明确一个前提pymysql 是纯 Python 实现的 MySQL 客户端不需要编译 C 扩展pip install pymysql就能用。它适合中小规模的数据操作如果你要处理每秒上万次的并发写入那得考虑其他方案。但对于本地脚本和小型服务pymysql 的易用性和可读性足够好。下面从环境准备开始一步步来。2. 用 TaoToken 统一 Key 通道pymysql 脚本里的 API 调用怎么收口在写数据库代码之前先说一下 API 调用统一走 TaoToken 这件事。很多人的脚本里会混着两种东西一种是直接连 MySQL 做增删改查另一种是调用外部 API 做数据补全、文本处理或者模型推理。后者的 Key 管理往往很乱——有的写在代码里有的放在环境变量有的用配置文件换一个项目就要重新配一遍。TaoToken 的思路是提供一个统一的 Key 通道你把 API 调用都指向同一个入口Key 只维护一份。具体到操作上你需要先拿到一个可用的 Key。访问 https://taotoken.net/api-keys 创建或查看你的 API Key然后把它放到环境变量里不要硬编码在脚本中。比如在 Linux 或 macOS 的终端里export TAOTOKEN_API_KEY你的KeyWindows 的话用set TAOTOKEN_API_KEY你的Key或者在系统环境变量里配置。这样你的 Python 脚本通过os.environ.get(TAOTOKEN_API_KEY)就能读到换机器或者换项目时只改环境变量不动代码。TaoToken 的 API 入口是 https://taotoken.net/api模型对话、Coding Plan、控制台这些功能都从这里进。如果你用的是 Claude Code 或者类似的编码工具可以在配置里把 Base URL 指向 TaoToken 的地址Model ID 按你实际使用的模型填。比如在 Claude Code 的配置里你需要填三样东西Base URL、API Key、Model ID。Base URL 用https://taotoken.net/apiAPI Key 用你刚创建的那个Model ID 根据你订阅的模型来写。这样你的编码工具和 Python 脚本就共用同一个 Key 通道不用来回切换。对于 pymysql 脚本来说API 调用和数据库操作是两条线。数据库这条线走本地 MySQL 连接API 这条线走 TaoToken。你可以在同一个脚本里同时用比如先从 MySQL 查出一批待处理的记录然后逐条调用 TaoToken 的模型对话接口做文本分类最后把结果写回 MySQL。这种模式在数据清洗和内容处理场景里很常见。关键是 Key 只维护一份数据库密码也只维护一份两边不混。如果你还没有 TaoToken 的 Key可以先注册一个账号然后在控制台里创建。整个过程不需要什么特殊配置拿到 Key 之后按上面的方式放到环境变量里就行。接下来进入 pymysql 的具体操作。3. 可复制配置pymysql 连接、建表、增删改查逐行注释这一节是核心所有代码都可以直接复制运行。我按“连接 → 建表 → 插入 → 查询 → 更新 → 删除”的顺序来每一行都加中文注释。你先确保本地 MySQL 已经启动并且有一个可用的数据库。如果没有可以用CREATE DATABASE guest DEFAULT CHARACTER SET utf8mb4;建一个。3.1 连接配置与建表import pymysql.cursors # 导入 pymysql 的游标模块DictCursor 让查询结果以字典返回 # 建立数据库连接 connection pymysql.connect( host127.0.0.1, # 数据库主机地址本机就是 127.0.0.1 port3306, # MySQL 默认端口没改过就填 3306 userroot, # 数据库用户名按你实际的填 password你的密码, # 数据库密码不要硬编码在正式项目里 dbguest, # 要连接的数据库名 charsetutf8mb4, # 通信编码必须和建库时一致否则中文乱码 cursorclasspymysql.cursors.DictCursor # 游标类型查询结果以字典形式返回 ) # 创建游标对象后续所有 SQL 都通过它执行 cursor connection.cursor() # 建表语句如果表已存在就跳过 create_table_sql CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, # 自增主键 email VARCHAR(255) NOT NULL, # 邮箱不允许为空 password VARCHAR(255) NOT NULL, # 密码实际项目要存哈希值 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP # 创建时间自动填充 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; # 引擎和字符集 cursor.execute(create_table_sql) # 执行建表语句 connection.commit() # 提交事务建表操作生效 print(表创建成功或已存在)这段代码里pymysql.connect()返回一个 Connection 对象它代表和数据库的一条连接。cursor()方法返回 Cursor 对象你所有的 SQL 都通过cursor.execute()来执行。commit()是必须的因为 pymysql 默认不会自动提交你不 commit插入和更新就不会真正写到数据库里。这一点和很多 ORM 框架不一样新手最容易在这里踩坑。3.2 插入数据# 插入一条用户记录 insert_sql INSERT INTO users (email, password) VALUES (%s, %s) # 用占位符防止 SQL 注入 cursor.execute(insert_sql, (huzhihengitest.info, 123456)) # 执行插入参数以元组传入 connection.commit() # 提交事务数据真正写入 print(插入成功受影响行数, cursor.rowcount) # rowcount 返回受影响的行数注意这里用了%s占位符而不是直接把变量拼到 SQL 字符串里。这是防止 SQL 注入的基本做法pymysql 会自动帮你转义参数。如果你写fINSERT ... VALUES ({email}, {password})遇到带引号的输入就会出问题严重的话会被注入攻击。养成用占位符的习惯后面查询和更新也一样。3.3 查询数据# 查询单条数据 select_one_sql SELECT id, password FROM users WHERE email%s # 按邮箱查 cursor.execute(select_one_sql, (huzhihengitest.info,)) # 注意单参数也要用元组末尾加逗号 result cursor.fetchone() # fetchone 返回结果集的第一行没有则返回 None print(单条查询结果, result) # DictCursor 下返回的是字典 # 查询多条数据 select_all_sql SELECT id, password FROM users # 查全表 cursor.execute(select_all_sql) # 执行查询 results cursor.fetchall() # fetchall 返回所有行每行是一个字典 for row in results: # 遍历结果集 print(多条查询结果, row) # 逐行打印fetchone()和fetchall()的区别很直观前者只取第一行后者取全部。如果你确定只需要一条记录用fetchone()更省内存。fetchall()会把所有结果加载到内存里表很大的时候要注意。另外游标是有位置的fetchone()取完第一行后再调一次fetchone()会取第二行。如果你想重新从头取可以用cursor.scroll(0, modeabsolute)把指针移回开头。3.4 更新与删除# 更新数据 update_sql UPDATE users SET password%s WHERE email%s # 按邮箱更新密码 cursor.execute(update_sql, (654321, huzhihengitest.info)) # 传入新密码和条件 connection.commit() # 提交更新 print(更新成功受影响行数, cursor.rowcount) # 删除数据 delete_sql DELETE FROM users WHERE email%s # 按邮箱删除 cursor.execute(delete_sql, (huzhihengitest.info,)) # 执行删除 connection.commit() # 提交删除 print(删除成功受影响行数, cursor.rowcount) # 关闭游标和连接 cursor.close() # 关闭游标释放资源 connection.close() # 关闭连接断开和数据库的会话更新和删除同样需要commit()不提交就不会生效。rowcount可以帮你确认到底影响了几行如果返回 0说明条件没匹配到任何记录。最后记得cursor.close()和connection.close()虽然 Python 的垃圾回收最终会处理但显式关闭更稳妥尤其是在循环里反复创建连接的时候。4. 验证请求一条查询确认连通与返回结果代码写完之后怎么确认真的连上了、数据真的写进去了最直接的办法就是跑一条查询看返回结果。你可以把上面的插入和查询串起来写一个完整的验证脚本import pymysql.cursors # 导入模块 # 连接数据库 connection pymysql.connect( host127.0.0.1, port3306, userroot, password你的密码, dbguest, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor ) cursor connection.cursor() # 创建游标 # 插入一条测试数据 cursor.execute(INSERT INTO users (email, password) VALUES (%s, %s), (testexample.com, abc123)) connection.commit() # 提交插入 # 查询刚插入的数据 cursor.execute(SELECT id, email, password FROM users WHERE email%s, (testexample.com,)) result cursor.fetchone() # 取一条 print(查询结果, result) # 打印结果 # 清理测试数据 cursor.execute(DELETE FROM users WHERE email%s, (testexample.com,)) connection.commit() # 提交删除 cursor.close() # 关闭游标 connection.close() # 关闭连接运行之后如果看到类似{id: 1, email: testexample.com, password: abc123}的输出说明连接、插入、查询、删除整条链路都通了。如果报错先看错误类型Access denied是用户名或密码不对Unknown database是数据库名写错Table doesnt exist是表没建Cant connect to MySQL server是 MySQL 没启动或者 host/port 不对。逐项排查基本都能解决。另外如果你在脚本里同时用了 TaoToken 的 API可以单独验证一下 API 通道。比如用requests库向https://taotoken.net/api发一个模型对话请求带上你的 Key看能不能正常返回。数据库和 API 两条线都验证通过整个脚本才算真正可用。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 怎么处理这一节列几个实际写代码时高频出现的报错以及对应的排查思路。你遇到问题可以先在这里对照。401 Unauthorized这个通常出现在 API 调用侧不是数据库侧。如果你用 TaoToken 的 Key 调模型接口返回 401先检查 Key 是不是复制错了有没有多余空格。然后确认请求头里的Authorization格式对不对一般是Bearer 你的Key。如果 Key 没问题检查环境变量有没有生效echo $TAOTOKEN_API_KEY看一下输出。数据库侧的 401 对应的是Access denied for user那是 MySQL 用户名密码的问题和 API 的 401 不是一回事。local proxy failed这个报错一般出现在网络请求环节说明请求没有正确到达目标地址。先确认你的 Base URL 写对了TaoToken 的 API 入口是https://taotoken.net/api不要多写或少写路径。然后检查本地网络能不能正常访问这个地址可以用curl -I https://taotoken.net/api试一下。如果公司网络有特殊限制可能需要换网络环境再试。注意不要配置任何非官方的网络工具直接用系统默认网络即可。reading choices 相关报错这个通常出现在解析模型返回结果的时候。如果你调模型对话接口返回的 JSON 里choices字段是空的或者格式不对先打印完整的响应内容看看。可能是请求参数里model字段填错了或者messages格式不对。TaoToken 的模型对话接口兼容常见的请求格式你按文档里的示例传参就行。如果返回的是流式数据解析方式和平常不一样要按流式的方式逐块读取。OAuth 相关报错如果你用 Claude Code 或者其他编码工具接入 TaoToken遇到 OAuth 报错先确认工具版本是不是太旧。然后检查配置里的 Base URL、API Key、Model ID 三件套是不是都填了。Base URL 用https://taotoken.net/apiAPI Key 用你在控制台创建的那个Model ID 按你实际订阅的模型写。三个缺一个都可能报 OAuth 错误。如果还是不行把工具的日志打开看具体是哪一步失败了。pymysql 侧的常见错pymysql.err.OperationalError: (2003, Cant connect to MySQL server on 127.0.0.1)说明 MySQL 没启动或者端口不对pymysql.err.ProgrammingError: (1146, Table guest.users doesnt exist)说明表没建pymysql.err.IntegrityError: (1062, Duplicate entry)说明主键或唯一索引冲突。这些错误信息都很直白按提示改就行。排查的时候有一个通用原则先确认连接层通不通再确认 SQL 对不对最后确认事务提交了没有。大部分问题都出在这三步里。6. 把 Key 和数据库密码管好脚本才能长期跑最后说一个实际维护的问题。你写脚本的时候数据库密码和 API Key 都不应该硬编码在代码里。数据库密码可以放在环境变量或者单独的配置文件里API Key 同样。TaoToken 的 Key 在控制台可以创建多个你可以给不同的项目分配不同的 Key方便追踪用量和随时吊销。如果某个 Key 泄露了直接删掉重新创建一个就行不影响其他项目。对于小型服务建议把数据库连接封装成一个函数每次需要的时候调用用完关闭。不要在全局维护一个长期不关的连接除非你用连接池。pymysql 本身不提供连接池但你可以用DBUtils这样的库来管理。不过对于本地脚本每次连接用完就关简单直接不容易出问题。如果你后续要把脚本部署到服务器上记得把环境变量也配好不要依赖本地的 shell 配置。可以用 systemd 的Environment指令或者在启动脚本里 export。数据库那边建议给脚本单独建一个数据库用户只授予必要的权限不要直接用 root。这样即使脚本出问题影响范围也可控。整套流程走下来你应该能独立写出带完整注释的 pymysql 操作代码并且知道 API 调用怎么统一走 TaoToken 的 Key 通道。遇到报错的时候按连接层、SQL 层、事务层的顺序排查大部分问题都能自己解决。