1. 银行转账场景下的事务边界到底卡在哪先把问题摆出来Python 连 MySQL 时connect()返回的连接对象、cursor()返回的游标对象、commit()和rollback()这几个方法很多人是「照着示例能跑换个场景就翻车」。最典型的翻车现场就是转账——A 账户扣 100B 账户加 100中间任何一步抛异常钱就凭空消失或者凭空多出来。我见过太多代码长这样先execute一条 UPDATE 扣钱再execute一条 UPDATE 加钱两条都成功了才想起来commit()。问题在于如果第二条 UPDATE 因为余额不足、字段类型不匹配、连接超时等原因失败第一条 UPDATE 已经在当前事务里生效了而你没有rollback()这个事务会一直挂着直到连接关闭时被隐式回滚或者被后续的commit()一起提交——两种结果都不是你想要的。这里要厘清一个核心概念MySQL 默认的存储引擎 InnoDB 是支持事务的而 Python 的 MySQL 驱动无论是 MySQLdb 还是 PyMySQL默认开启了 autocommit0。也就是说你执行的每一条 INSERT/UPDATE/DELETE 都处在一个隐式开启的事务里只有显式调用commit()才会真正落盘调用rollback()才会撤销。cursor()则是你操作 SQL 的「手柄」所有execute、executemany、fetchone、fetchall都挂在游标上而不是连接上。所以转账场景的正确姿势是一个连接、一个游标、两条 UPDATE、一次 commit异常时一次 rollback。听起来简单但参数怎么配、异常怎么捕、回滚后怎么验证数据一致性每一步都有坑。下面我把connect()的参数配置、游标写法、事务提交与回滚的完整链路拆开讲代码可以直接复制去跑。这一节先明确适用人群如果你正在写 Python 后端、做数据同步脚本、或者维护老项目里的 MySQLdb 代码这篇内容能帮你把事务边界控制这件事彻底理清。如果你只是偶尔查个数据那cursor()的 fetch 系列方法也够你用。2. TaoToken 前置把模型对话和 API Key 准备好在写代码之前有个前置动作值得先做——把调试过程中需要用到的模型对话能力和 API Key 准备好。我自己的习惯是遇到 SQL 报错或者事务行为不符合预期时直接把报错信息和表结构丢给模型对话让它帮我分析是隔离级别问题还是驱动参数问题比翻文档快很多。TaoToken 的模型对话入口在 https://taotoken.net/api API Key 在 https://taotoken.net/api-keys 生成。如果你打算长期写代码、跑 Agent 任务Coding Plan 的入口在 https://taotoken.net/coding-plan 适合把模型能力嵌进日常开发流里。接入文档在 https://taotoken.net/doc 里面有各语言 SDK 的调用示例。具体操作上你先生成一个 API Key然后拿这个 Key 去调模型对话接口。比如你想让模型帮你检查一段事务代码有没有漏掉 rollback可以把代码贴进去问「这段代码在第二条 UPDATE 失败时第一条 UPDATE 会不会被提交」。模型对话的 Base URL 用https://taotoken.net/apiKey 用你生成的那串Model ID 按文档里列的填。这里要提醒一句TaoToken 是模型调用平台不是数据库工具它不会帮你连 MySQL。它的作用是当你在写connect()参数、调commit()时机、排查rollback()没生效这类问题时给你一个能快速问答的助手。真正连数据库、执行 SQL 的还是你本地的 Python 环境和 MySQL 服务。如果你用的是 Claude Code 这类编码工具接入方式是把 Base URL 和 Key 配到对应的配置文件里。以 Claude Code 为例它的配置走的是 Anthropic 兼容接口Base URL 填https://taotoken.net/apiKey 填你生成的Model ID 按文档选。配好之后你在终端里写代码遇到事务相关的问题可以直接问不用切浏览器。Cline 的 MCP 配置也是类似的思路在 MCP 配置文件里加上 TaoToken 的接入信息Base URL、Key、Model ID 三件套填全。Codex 的 auth.json 同理把这三个字段写进去就能用。这三件套缺一不可尤其是 Model ID填错了会直接报模型不存在。把这一步做完你后面写代码时遇到commit()没生效、rollback()报错、游标 fetch 结果不对这些问题就有一个即时的问答通道。接下来进入正题开始配connect()。3. 可复制的 connect() 参数配置与游标写法先给一份可以直接跑的配置。我用的是 PyMySQL因为它是纯 Python 实现装起来比 MySQLdb 省事API 也基本兼容。如果你项目里用的是 MySQLdb把 import 换掉即可参数名一致。import pymysql conn pymysql.connect( host127.0.0.1, port3306, userroot, passwordyour_password, databasebank_demo, charsetutf8mb4, autocommitFalse, cursorclasspymysql.cursors.Cursor, connect_timeout10, read_timeout30, write_timeout30, )逐项说明host和port是 MySQL 服务地址本地就是 127.0.0.1:3306。user和password是你的账号密码。database指定默认库省得每条 SQL 都写库名。charset建议用 utf8mb4能存 emoji 和生僻字。autocommitFalse是关键显式关闭自动提交这样你才能手动控制事务边界。cursorclass默认是元组游标如果你想要字典游标结果按列名取值改成pymysql.cursors.DictCursor。三个 timeout 参数建议都加上避免网络抖动时程序卡死。连接建好后游标这样拿cursor conn.cursor()cursor()方法本身可以传参数比如conn.cursor(pymysql.cursors.DictCursor)可以覆盖连接级别的游标类型。游标对象上挂着你所有要用的方法execute(sql, args)执行单条executemany(sql, seq_of_args)批量执行fetchone()取一行fetchmany(size)取 size 行fetchall()取全部scroll(value, mode)移动指针。转账场景的建表语句先准备好CREATE TABLE accounts ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(50) NOT NULL, balance DECIMAL(10,2) NOT NULL DEFAULT 0.00 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; INSERT INTO accounts (name, balance) VALUES (Alice, 1000.00), (Bob, 500.00);注意引擎必须是 InnoDBMyISAM 不支持事务rollback()会静默失效。这是很多人踩过的坑表建好了代码也写了 rollback但数据还是变了一查引擎是 MyISAM。现在写转账的核心逻辑。先看正常提交的版本def transfer(conn, from_id, to_id, amount): cursor conn.cursor() try: cursor.execute( UPDATE accounts SET balance balance - %s WHERE id %s, (amount, from_id) ) cursor.execute( UPDATE accounts SET balance balance %s WHERE id %s, (amount, to_id) ) conn.commit() print(转账成功已提交) except Exception as e: conn.rollback() print(f转账失败已回滚: {e}) finally: cursor.close()这段代码里两条 UPDATE 在同一个事务里commit()之前都没有真正落盘。如果第二条 UPDATE 抛异常except分支执行rollback()第一条 UPDATE 的扣款被撤销数据回到转账前的状态。finally里关游标但注意不要在这里关连接连接通常复用。再看异常回滚的验证版本故意制造一个错误def transfer_with_error(conn, from_id, to_id, amount): cursor conn.cursor() try: cursor.execute( UPDATE accounts SET balance balance - %s WHERE id %s, (amount, from_id) ) # 故意写错表名触发异常 cursor.execute( UPDATE account_wrong SET balance balance %s WHERE id %s, (amount, to_id) ) conn.commit() except Exception as e: conn.rollback() print(f捕获异常: {e}) finally: cursor.close()跑完之后去查 Alice 的余额应该还是 1000.00没有被扣掉。这就是rollback()的作用。批量插入用executemany比循环execute快很多def batch_insert(conn, rows): cursor conn.cursor() try: cursor.executemany( INSERT INTO accounts (name, balance) VALUES (%s, %s), rows ) conn.commit() print(f批量插入 {cursor.rowcount} 行) except Exception as e: conn.rollback() print(f批量插入失败: {e}) finally: cursor.close()executemany的第二个参数是一个序列每个元素是一个元组对应 SQL 里的占位符。cursor.rowcount返回受影响行数。注意executemany本身不提交还是要commit()。关于占位符PyMySQL 和 MySQLdb 都用%s不是?。这是从 SQLite 转过来的人最容易写错的地方写成?会直接报语法错误。4. 验证请求与成功结果commit 和 rollback 的实际表现代码写完了怎么确认commit()真的提交了、rollback()真的回滚了不能只看 print 输出要去数据库里查实际数据。先准备一个验证脚本每次操作前后都查一次余额def check_balance(conn, account_id): cursor conn.cursor() cursor.execute(SELECT balance FROM accounts WHERE id %s, (account_id,)) row cursor.fetchone() cursor.close() return row[0] if row else None然后跑一个完整的验证流程conn pymysql.connect( host127.0.0.1, port3306, userroot, passwordyour_password, databasebank_demo, charsetutf8mb4, autocommitFalse ) print(初始 Alice:, check_balance(conn, 1)) print(初始 Bob:, check_balance(conn, 2)) # 正常转账 transfer(conn, 1, 2, 100) print(转账后 Alice:, check_balance(conn, 1)) print(转账后 Bob:, check_balance(conn, 2)) # 异常转账 transfer_with_error(conn, 1, 2, 200) print(异常后 Alice:, check_balance(conn, 1)) print(异常后 Bob:, check_balance(conn, 2))预期输出初始 Alice 1000、Bob 500正常转账后 Alice 900、Bob 600异常转账后 Alice 还是 900、Bob 还是 600因为回滚了。这里有个细节要注意check_balance里用了新的游标但用的是同一个连接。因为autocommitFalse这个查询也在一个事务里看到的是当前事务的快照。如果你在rollback()之后立刻查看到的是回滚后的数据如果你在commit()之前查看到的是未提交的数据同一个连接内可见。跨连接的话未提交的数据对其他连接不可见这是事务隔离级别决定的。再验证一下executemany的批量插入rows [(Charlie, 300.00), (David, 400.00), (Eve, 500.00)] batch_insert(conn, rows) cursor conn.cursor() cursor.execute(SELECT COUNT(*) FROM accounts) print(总账户数:, cursor.fetchone()[0]) cursor.close()预期总账户数是 5Alice、Bob 加三个新的。如果executemany中途失败rollback()会把整批插入都撤销不会出现插了一半的情况。游标的 fetch 系列方法也验证一下cursor conn.cursor() cursor.execute(SELECT id, name, balance FROM accounts ORDER BY id) print(fetchone:, cursor.fetchone()) print(fetchmany(2):, cursor.fetchmany(2)) print(fetchall:, cursor.fetchall()) cursor.close()fetchone取第一行fetchmany(2)取接下来两行fetchall取剩下的。游标指针会随着 fetch 移动取过的不会再取。如果想重新从头取用cursor.scroll(0, modeabsolute)把指针移回开头。scroll的用法cursor conn.cursor() cursor.execute(SELECT id, name FROM accounts ORDER BY id) cursor.scroll(2, modeabsolute) # 移到第 3 行索引从 0 开始 print(第 3 行:, cursor.fetchone()) cursor.scroll(-1, moderelative) # 从当前位置往回移 1 行 print(回退 1 行:, cursor.fetchone()) cursor.close()modeabsolute是从结果集第一行开始算moderelative是从当前位置算。负数表示往回移。跑完这些验证你应该能确认commit()之后数据落盘rollback()之后数据撤销cursor()的 fetch 和 scroll 行为符合预期。接下来看常见报错。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节把写代码和调模型时最容易撞上的几类报错集中说一下。第一类数据库连接报错。最常见的是pymysql.err.OperationalError: (2003, Cant connect to MySQL server on 127.0.0.1)说明 MySQL 服务没启动或者端口不对。先确认systemctl status mysql或者netstat -tlnp | grep 3306。如果是(1045, Access denied for user rootlocalhost)是密码错了或者用户权限不对检查connect()里的 user 和 password。如果是(1049, Unknown database bank_demo)是库没建先CREATE DATABASE bank_demo。第二类事务没生效。代码里写了rollback()但数据还是变了九成是表引擎不是 InnoDB。用SHOW TABLE STATUS LIKE accounts查 Engine 字段如果是 MyISAM改成 InnoDBALTER TABLE accounts ENGINEInnoDB。另一个可能是autocommitTrue这样每条 SQL 都自动提交rollback()无从下手。检查connect()里有没有显式设autocommitFalse。第三类模型调用报 401。这个在配 TaoToken 的时候会遇到。401 的意思是认证失败通常是 API Key 没填对、填了但过期了、或者 Key 和 Base URL 不匹配。检查你的 Key 是不是从 https://taotoken.net/api-keys 生成的Base URL 是不是https://taotoken.net/api。如果用的是 Claude Code 或 Cline检查配置文件里的字段名有没有写错比如把api_key写成了apikey。第四类local proxy failed。这个报错通常出现在本地网络环境有代理设置的时候。模型调用请求发不出去客户端报 local proxy failed。检查你的环境变量HTTP_PROXY、HTTPS_PROXY有没有设成不可用的地址或者系统代理有没有开。把代理关掉或者把 TaoToken 的域名加到代理白名单里。注意这里说的是本地网络配置问题不是让你去搞什么特殊网络工具就是把系统代理设置检查一遍。第五类reading choices 报错。这个一般出现在模型返回的 JSON 结构不符合预期时。比如你期望返回choices[0].message.content但实际返回的结构里没有choices字段或者choices是空数组。先打印完整响应体看看结构确认 Model ID 填对了、请求参数里的messages格式正确。如果用的是 OpenAI 兼容接口检查model字段是不是文档里列的有效值。第六类OAuth 相关报错。如果你用的是需要 OAuth 授权的编码工具报 OAuth 失败通常是 token 过期或者回调地址不对。重新走一遍授权流程确认回调地址和工具里配的一致。有些工具把 OAuth token 缓存在本地文件里删掉缓存文件重新授权也能解决。第七类游标相关报错。pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax)多半是占位符写成了?或者 SQL 拼写错误。pymysql.err.InterfaceError: (0, )通常是游标已经关闭了还在用检查cursor.close()之后有没有再调execute。pymysql.err.InternalError: (1366, Incorrect string value)是字符集问题把charset改成utf8mb4。第八类连接池耗尽。如果你在循环里反复connect()而不close()连接数会涨到 MySQL 的max_connections上限报Too many connections。正确做法是复用一个连接或者用连接池比如 DBUtils 的 PooledDB。每次操作完关游标但连接留着复用。把这几类报错对照着排查基本能覆盖 90% 的常见问题。如果遇到没列出来的把完整报错信息丢给模型对话让它帮你定位。6. 语义一致 CTA把调试通道和编码计划配好代码跑通之后建议把调试通道固定下来。我自己的做法是本地开一个终端跑 Python 脚本浏览器开着模型对话页面遇到报错直接复制粘贴过去问。模型对话入口在 https://taotoken.net/api API Key 在 https://taotoken.net/api-keys 生成接入文档在 https://taotoken.net/doc 有各语言的调用示例。如果你打算长期写代码、跑 Agent 任务Coding Plan 的入口在 https://taotoken.net/coding-plan 适合把模型能力嵌进日常开发流。控制台在 https://taotoken.net/console 可以看调用量和余额。Claude Code 的接入配置Base URL 填https://taotoken.net/apiKey 填生成的 API KeyModel ID 按文档选。Cline 的 MCP 配置同理三件套填全。Codex 的 auth.json 里把这三个字段写进去。配好之后你在终端里写代码遇到commit()时机不对、rollback()没生效、游标 fetch 结果异常这些问题可以直接问不用切浏览器。最后留一个实用技巧转账这类涉及多表更新的操作建议在commit()之前加一条SELECT确认中间状态比如扣款后先查一次余额确认扣对了再加钱。这样即使逻辑有 bug也能在提交前发现。另外生产环境建议把connect_timeout、read_timeout、write_timeout都设上避免网络抖动导致连接挂死。事务里的 SQL 尽量短平快不要在事务里做耗时操作否则会长时间持锁影响其他连接。