1. 银行转账场景下的事务边界到底卡在哪
先把问题摆出来:Python 连 MySQL 时,connect()返回的连接对象、cursor()返回的游标对象、commit()和rollback()这几个方法,很多人是「照着示例能跑,换个场景就翻车」。最典型的翻车现场就是转账——A 账户扣 100,B 账户加 100,中间任何一步抛异常,钱就凭空消失或者凭空多出来。
我见过太多代码长这样:先execute一条 UPDATE 扣钱,再execute一条 UPDATE 加钱,两条都成功了才想起来commit()。问题在于,如果第二条 UPDATE 因为余额不足、字段类型不匹配、连接超时等原因失败,第一条 UPDATE 已经在当前事务里生效了,而你没有rollback(),这个事务会一直挂着,直到连接关闭时被隐式回滚或者被后续的commit()一起提交——两种结果都不是你想要的。
这里要厘清一个核心概念:MySQL 默认的存储引擎 InnoDB 是支持事务的,而 Python 的 MySQL 驱动(无论是 MySQLdb 还是 PyMySQL)默认开启了 autocommit=0。也就是说,你执行的每一条 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/api,Key 用你生成的那串,Model ID 按文档里列的填。
这里要提醒一句:TaoToken 是模型调用平台,不是数据库工具,它不会帮你连 MySQL。它的作用是当你在写connect()参数、调commit()时机、排查rollback()没生效这类问题时,给你一个能快速问答的助手。真正连数据库、执行 SQL 的还是你本地的 Python 环境和 MySQL 服务。
如果你用的是 Claude Code 这类编码工具,接入方式是把 Base URL 和 Key 配到对应的配置文件里。以 Claude Code 为例,它的配置走的是 Anthropic 兼容接口,Base URL 填https://taotoken.net/api,Key 填你生成的,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( host="127.0.0.1", port=3306, user="root", password="your_password", database="bank_demo", charset="utf8mb4", autocommit=False, cursorclass=pymysql.cursors.Cursor, connect_timeout=10, read_timeout=30, write_timeout=30, )逐项说明:host和port是 MySQL 服务地址,本地就是 127.0.0.1:3306。user和password是你的账号密码。database指定默认库,省得每条 SQL 都写库名。charset建议用 utf8mb4,能存 emoji 和生僻字。autocommit=False是关键,显式关闭自动提交,这样你才能手动控制事务边界。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 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; INSERT INTO accounts (name, balance) VALUES ('Alice', 1000.00), ('Bob', 500.00);注意引擎必须是 InnoDB,MyISAM 不支持事务,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( host="127.0.0.1", port=3306, user="root", password="your_password", database="bank_demo", charset="utf8mb4", autocommit=False ) 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里用了新的游标,但用的是同一个连接。因为autocommit=False,这个查询也在一个事务里,看到的是当前事务的快照。如果你在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()预期总账户数是 5(Alice、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, mode='absolute')把指针移回开头。
scroll的用法:
cursor = conn.cursor() cursor.execute("SELECT id, name FROM accounts ORDER BY id") cursor.scroll(2, mode='absolute') # 移到第 3 行(索引从 0 开始) print("第 3 行:", cursor.fetchone()) cursor.scroll(-1, mode='relative') # 从当前位置往回移 1 行 print("回退 1 行:", cursor.fetchone()) cursor.close()mode='absolute'是从结果集第一行开始算,mode='relative'是从当前位置算。负数表示往回移。
跑完这些验证,你应该能确认:commit()之后数据落盘,rollback()之后数据撤销,cursor()的 fetch 和 scroll 行为符合预期。接下来看常见报错。
5. 本篇常见错排查:401、local proxy failed、reading choices、OAuth
这一节把写代码和调模型时最容易撞上的几类报错集中说一下。
第一类:数据库连接报错。最常见的是pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on '127.0.0.1'"),说明 MySQL 服务没启动或者端口不对。先确认systemctl status mysql或者netstat -tlnp | grep 3306。如果是(1045, "Access denied for user 'root'@'localhost'"),是密码错了或者用户权限不对,检查connect()里的 user 和 password。如果是(1049, "Unknown database 'bank_demo'"),是库没建,先CREATE DATABASE bank_demo。
第二类:事务没生效。代码里写了rollback()但数据还是变了,九成是表引擎不是 InnoDB。用SHOW TABLE STATUS LIKE 'accounts'查 Engine 字段,如果是 MyISAM,改成 InnoDB:ALTER TABLE accounts ENGINE=InnoDB。另一个可能是autocommit=True,这样每条 SQL 都自动提交,rollback()无从下手。检查connect()里有没有显式设autocommit=False。
第三类:模型调用报 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/api,Key 填生成的 API Key,Model ID 按文档选。Cline 的 MCP 配置同理,三件套填全。Codex 的 auth.json 里把这三个字段写进去。配好之后,你在终端里写代码,遇到commit()时机不对、rollback()没生效、游标 fetch 结果异常这些问题,可以直接问,不用切浏览器。
最后留一个实用技巧:转账这类涉及多表更新的操作,建议在commit()之前加一条SELECT确认中间状态,比如扣款后先查一次余额,确认扣对了再加钱。这样即使逻辑有 bug,也能在提交前发现。另外,生产环境建议把connect_timeout、read_timeout、write_timeout都设上,避免网络抖动导致连接挂死。事务里的 SQL 尽量短平快,不要在事务里做耗时操作,否则会长时间持锁,影响其他连接。