1. 为什么你的 pymysql try 总是“看起来没报错数据却丢了”先说结论pymysql的try不是把代码包起来就完事它要解决的是三件事——连接是否建立、SQL 是否执行成功、事务是否提交或回滚。很多人写try/except只捕获了异常却忘了commit和rollback的边界结果就是“程序没崩但表没建、数据没写”。我见过最常见的三种翻车现场第一种连接阶段就失败。pymysql.connect()抛的是OperationalError比如账号密码错、端口不通、库不存在。如果你把connect写在try外面程序直接崩连except都进不去。第二种SQL 语法错误。cursor.execute()抛ProgrammingError比如表名写错、字段类型不匹配。这时候事务已经开了如果不rollback连接池里的连接会一直挂着未提交事务后续请求可能读到脏数据。第三种最隐蔽的——commit没执行。pymysql默认autocommitFalse你execute完不commit数据只在当前连接可见关掉连接就没了。很多人以为try里执行成功就完事其实还差最后一步。所以这篇不讲虚的直接给你一套能复制的连接配置、异常捕获模板再演示一次“故意写错 SQL → 捕获报错 → 回滚 → 修复验证”的完整动作。适合刚用pymysql做脚本、爬虫入库、后台小工具的同学。核心检索词就三个python、pymysql、sql加上try的异常处理写法。你可能会问用with上下文管理器不是更省事吗pymysql的Connection支持with但它只管关闭连接不管commit。事务提交还是得你自己写。所以try/except/finally依然是绕不开的基本功。下面按“先能跑通再能排错”的顺序来。每一步都给完整代码你可以直接贴到本地改参数。2. TaoToken 前置把模型对话和 API Key 准备好边写边验证写pymysql的异常处理最烦的是“报错信息看不懂”。比如pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax...)后面跟一大串新手容易懵。这时候有个能随时问的模型对话入口效率会高很多。我平时会把报错原文贴到模型对话里让它帮我定位是 SQL 语法问题还是参数绑定问题。TaoToken 的模型对话入口在这里模型对话https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel_chat如果你要长期写代码、跑 Agent 任务可以看 Coding Plan它更适合连续编码场景Coding Planhttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding_planAPI Key 在控制台生成路径是 console → api-keys控制台https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentconsole API Keyshttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi_keys接入文档在这里里面有 Base URL 和 Model ID 的说明接入文档https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc如果你用的是 Claude Code 这类工具Anthropic 兼容入口也准备好了ClaudeCodeAnthropichttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentclaude_code官网首页https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content注意API 地址是https://taotoken.net/api不带 UTM 参数配置的时候别写错。为什么要在写pymysql之前提这个因为异常处理的核心是“看懂报错”。pymysql的异常类有十几种OperationalError、ProgrammingError、IntegrityError、InterfaceError每种对应的处理动作不一样。你把这些报错贴给模型让它帮你判断该rollback还是该重连比你自己翻文档快。举个例子IntegrityError是唯一键冲突这种错误rollback之后可以重试但OperationalError如果是连接断了rollback没用得重新connect。这个判断逻辑模型对话能帮你快速理清。准备好 Key 之后我们进入正题。先给一套最小可用的连接配置。3. 可复制配置pymysql 连接参数与 try/except/finally 模板这一节给你两份东西一份是连接配置的 JSON 片段方便你放到配置文件里一份是完整的try/except/finally代码模板。先说连接参数。pymysql.connect()常用参数如下参数说明建议值host数据库地址127.0.0.1port端口3306user用户名你的账号password密码你的密码database库名test_dbcharset字符集utf8mb4autocommit自动提交Falseconnect_timeout连接超时秒10cursorclass游标类型DictCursor配置文件建议写成 JSON路径放在项目根目录的config/db.json{ host: 127.0.0.1, port: 3306, user: testuser, password: testpassword, database: test_db, charset: utf8mb4, autocommit: false, connect_timeout: 10 }读取配置的代码import json import pymysql with open(config/db.json, r, encodingutf-8) as f: db_conf json.load(f) def get_conn(): return pymysql.connect(**db_conf)然后是核心的try/except/finally模板。注意三个关键点connect放在try里、commit放在execute之后、rollback放在except里、close放在finally里。import pymysql from pymysql import err def create_table(): conn None cursor None try: conn get_conn() cursor conn.cursor() my_sql CREATE TABLE IF NOT EXISTS test_db.new_table( col_1 VARCHAR(50) NOT NULL ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; cursor.execute(my_sql) conn.commit() print(创建成功) except err.ProgrammingError as e: if conn: conn.rollback() print(fSQL 语法错误已回滚: {e}) except err.OperationalError as e: if conn: conn.rollback() print(f连接或操作错误已回滚: {e}) except err.IntegrityError as e: if conn: conn.rollback() print(f数据完整性错误已回滚: {e}) except Exception as e: if conn: conn.rollback() print(f未知错误已回滚: {e}) finally: if cursor: cursor.close() if conn: conn.close() if __name__ __main__: create_table()这段代码和网上很多版本的区别在于conn和cursor初始化为Nonefinally里做判空。因为如果connect就失败了conn是None你直接conn.close()会再抛一个AttributeError把真正的报错盖掉。另外except里按异常类型分开写是为了让你知道到底哪类错误。如果你只想快速跑通写一个except Exception as e也行但排错时会丢信息。还有一个细节rollback只在conn存在时调用。如果连接都没建立rollback没意义。这套模板可以直接用在建表、插入、更新、删除。下面我们用一个插入场景来验证。4. 验证请求故意写错 SQL看 rollback 是否生效光看代码不够我们做一次完整的“报错复现 → 捕获 → 回滚 → 修复验证”。第一步建一张测试表CREATE TABLE IF NOT EXISTS test_db.user_info( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(50) NOT NULL, age INT NOT NULL ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;第二步写一个故意出错的插入函数。我们把表名写成user_info_wrong触发ProgrammingErrordef insert_user_wrong(): conn None cursor None try: conn get_conn() cursor conn.cursor() sql INSERT INTO test_db.user_info_wrong (name, age) VALUES (%s, %s) cursor.execute(sql, (张三, 25)) conn.commit() print(插入成功) except err.ProgrammingError as e: if conn: conn.rollback() print(f捕获到 SQL 错误: {e}) finally: if cursor: cursor.close() if conn: conn.close() insert_user_wrong()运行结果捕获到 SQL 错误: (1146, Table test_db.user_info_wrong doesnt exist)注意这里报的是1146表不存在属于ProgrammingError。rollback执行了连接关闭了程序没崩。第三步验证回滚是否真的生效。我们查一下user_info表里有没有“张三”def check_user(): conn get_conn() cursor conn.cursor() cursor.execute(SELECT COUNT(*) FROM test_db.user_info WHERE name %s, (张三,)) count cursor.fetchone()[0] cursor.close() conn.close() print(f张三的记录数: {count}) check_user()输出张三的记录数: 0说明回滚生效了错误的事务没有留下脏数据。第四步修复 SQL把表名改对def insert_user_right(): conn None cursor None try: conn get_conn() cursor conn.cursor() sql INSERT INTO test_db.user_info (name, age) VALUES (%s, %s) cursor.execute(sql, (张三, 25)) conn.commit() print(插入成功) except Exception as e: if conn: conn.rollback() print(f插入失败: {e}) finally: if cursor: cursor.close() if conn: conn.close() insert_user_right() check_user()输出插入成功 张三的记录数: 1到这里一次完整的“报错 → 回滚 → 修复 → 验证”就跑通了。你可以把insert_user_wrong和insert_user_right放在同一个文件里先跑错的再跑对的观察数据库变化。如果你在验证过程中遇到报错看不懂可以把报错贴到模型对话里模型对话https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel_chat5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节对照真实报错逐个说清楚原因和动作。注意这些报错分两类一类是pymysql本身的一类是你用模型对话或 API 时遇到的。两类都会影响你写代码的效率所以一起讲。报错一pymysql.err.OperationalError: (1045, Access denied for user...)这是账号密码错。检查config/db.json里的user和password。如果你用的是 Docker 里的 MySQL注意host不能写localhost要写127.0.0.1或者容器名。报错二pymysql.err.OperationalError: (2003, Cant connect to MySQL server on 127.0.0.1)端口不通。先确认 MySQL 服务起了没再确认port是不是 3306。如果是云数据库检查白名单有没有放你的 IP。报错三pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax)SQL 语法错。常见原因是表名或字段名用了 MySQL 保留字比如order、key要用反引号包起来。另外%s占位符不要加引号cursor.execute(sql, (张三, 25))这样传参pymysql会自动转义。报错四pymysql.err.IntegrityError: (1062, Duplicate entry xxx for key PRIMARY)唯一键冲突。这种错误rollback之后可以重试或者用INSERT ... ON DUPLICATE KEY UPDATE。报错五401 Unauthorized这是 API Key 的问题。检查你的 Key 有没有复制完整有没有多余空格。如果用的是 TaoToken去控制台重新生成一个API Keyshttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi_keys报错六local proxy failed本地代理配置问题。如果你在代码里设置了http_proxy或https_proxy环境变量但代理服务没起就会报这个。检查环境变量或者临时unset掉。报错七reading choices相关报错这通常出现在调用模型接口时返回体里没有choices字段。原因可能是请求格式不对或者 Model ID 写错了。对照接入文档检查接入文档https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc报错八OAuth相关报错如果你用 Claude Code 或类似工具OAuth 回调失败通常是回调地址不对。检查工具里的 Base URL 和 Key 配置。Claude Code 的 Anthropic 兼容入口ClaudeCodeAnthropichttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentclaude_code如果你用 CC Switch、Cline MCP 或 Codex 的auth.json配置要写全三件套Base URL、Key、Model ID。缺一个都会报错。配置项值Base URLhttps://taotoken.net/apiKey控制台生成的 API KeyModel ID接入文档里查到的模型名auth.json示例{ base_url: https://taotoken.net/api, api_key: sk-你的Key, model: 你的ModelID }注意Base URL 不要加 UTM 参数就写https://taotoken.net/api。报错九事务未提交数据查不到这不是报错是“静默失败”。pymysql默认autocommitFalse你execute完不commit数据只在当前连接可见。解决办法要么在connect时设autocommitTrue要么在execute后显式conn.commit()。推荐后者因为事务边界更清晰。报错十cursor或conn为None时调用close()这是finally里没判空导致的。按第 3 节的模板conn和cursor初始化为Nonefinally里判空再关。排错的核心思路是先看异常类型再看错误码最后看是连接层、SQL 层还是事务层。把这三层分开大部分问题都能定位。6. 语义一致 CTA把异常处理写成习惯而不是补丁写pymysql的try本质上是在给数据库操作划边界。连接是边界事务是边界资源释放也是边界。你把这三条边界画清楚代码就不会“看起来没报错数据却丢了”。我自己的习惯是每个操作数据库的函数都套用第 3 节的模板。connect在try里commit在execute后rollback在except里close在finally里。写多了就是肌肉记忆。如果你要长期写这类代码或者跑 Agent 任务自动入库Coding Plan 会更顺手Coding Planhttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding_plan遇到报错看不懂就去模型对话里贴原文模型对话https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel_chatKey 在控制台生成别忘了控制台https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentconsole最后留一个实用技巧在except里打印异常时用repr(e)而不是str(e)因为repr会带上错误码比如(1146, Table ... doesnt exist)而str可能只给你半句话。这个细节在排查pymysql报错时特别有用。