用 Python + MySQL 搭建模拟银行转账系统：TaoToken 统一 Key 配置与事务验证实践

1. 从一次转账失败说起为什么你的 Python MySQL 转账代码会丢钱模拟银行转账系统几乎是每个学 Python 操作 MySQL 的人都会碰到的练手项目建一张账户表写两条记录让账户 1 给账户 2 转 200 元跑通就完事。但真正把它当成一个系统来写问题立刻冒出来——转账到一半程序崩了钱扣了但没到账两个人同时从同一个账户扣款余额被扣成负数commit忘了写数据库里数据纹丝不动。这篇就围绕 Python MySQL 实现银行转账的核心流程展开覆盖账户表设计、事务提交与回滚、并发扣款异常处理给出可复制的config.toml与settings.json骨架并演示通过 TaoToken 统一 Key/API 通道接入 AI 辅助编码最后附上转账成功与余额不足两类验证动作。适合已经会写基础 SQL、想搞清楚事务到底怎么保证一致性的开发者也适合想把 AI 编码助手接进日常开发流的人。我试过用最朴素的写法跑转账结果并发一压就出问题后来才把事务隔离、行锁、异常回滚这几块补齐。下面按建表 → 配置 → 写代码 → 验证 → 排障的顺序走一遍每一步都能直接复制。2. 前置准备用 TaoToken 统一 Key 打通 AI 辅助编码通道写这类涉及事务和并发的代码最容易卡在细节上cursor.rowcount到底返回什么、rollback该放在哪一层、SELECT ... FOR UPDATE加在哪。与其反复查文档不如把 AI 编码助手接进来让它边写边解释。TaoToken 提供统一的 Key 和 API 通道一个 Key 就能调用多种模型省去在多个平台之间来回切换配置的麻烦。你需要先拿到一个 API Key。登录 TaoToken 控制台在 API Keys 页面创建一个新 Key复制保存好。这个 Key 后面会写进config.toml供本地脚本和 AI 编码工具共用。控制台入口https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentconsoleAPI Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi-keys接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdocAPI 基础地址统一用https://taotoken.net/api注意这个地址不带任何查询参数配置时直接填即可。如果你打算长期用 AI 辅助写代码、跑 Agent 任务可以了解下 Coding Plan它更适合高频编码场景Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding-plan注意API Key 属于敏感凭证不要硬编码进提交到 Git 的脚本里。下面我会把它放进config.toml并配合.gitignore排除。3. 可复制配置config.toml 与 settings.json 骨架先把配置骨架搭好。项目根目录建两个文件config.toml放数据库和 API 连接信息settings.json放运行时参数比如转账限额、重试次数。这样代码里读配置不用改一行逻辑就能切换环境。3.1 config.toml数据库与 TaoToken 连接# config.toml [mysql] host 127.0.0.1 port 3306 user root password your_mysql_password database bank_demo charset utf8mb4 [taotoken] api_base https://taotoken.net/api api_key sk-你的TaoToken密钥 model claude-sonnet-4-20250514 timeout 60 [transfer] max_amount 50000 # 单笔转账上限 max_retry 3 # 死锁重试次数 retry_interval 0.2 # 重试间隔秒3.2 settings.json运行时参数{ transfer: { max_amount: 50000, max_retry: 3, retry_interval: 0.2, isolation_level: REPEATABLE READ }, logging: { level: INFO, file: transfer.log }, ai_assist: { enabled: true, api_base: https://taotoken.net/api, model: claude-sonnet-4-20250514 } }3.3 账户表设计别再用 int 存金额原示例里用int(11)存金额实际项目里这是坑。金额用DECIMAL(15,2)避免浮点误差加一个version字段做乐观锁或者直接用SELECT ... FOR UPDATE做悲观锁。建表语句如下CREATE DATABASE IF NOT EXISTS bank_demo DEFAULT CHARSET utf8mb4; USE bank_demo; CREATE TABLE account ( acctid BIGINT NOT NULL PRIMARY KEY AUTO_INCREMENT, name VARCHAR(64) NOT NULL, money DECIMAL(15,2) NOT NULL DEFAULT 0.00, version INT NOT NULL DEFAULT 0, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_name (name) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; INSERT INTO account (acctid, name, money) VALUES (1, alice, 1000.00), (2, bob, 500.00);提示引擎必须是 InnoDBMyISAM 不支持事务rollback会失效。这是新手最常踩的坑之一。4. 核心实现事务提交、回滚与并发扣款配置就绪后进入代码部分。整体思路是一个TransferService类封装转账逻辑用SELECT ... FOR UPDATE锁定两行账户按固定顺序加锁避免死锁扣款和入账在同一个事务里任何一步异常就rollback。4.1 读取配置与建立连接# db.py import tomllib import pymysql from dbutils.pooled_db import PooledDB with open(config.toml, rb) as f: cfg tomllib.load(f) mysql_cfg cfg[mysql] _pool PooledDB( creatorpymysql, maxconnections10, mincached2, hostmysql_cfg[host], portmysql_cfg[port], usermysql_cfg[user], passwordmysql_cfg[password], databasemysql_cfg[database], charsetmysql_cfg[charset], autocommitFalse, cursorclasspymysql.cursors.DictCursor, ) def get_conn(): return _pool.connection()用连接池而不是每次Connect是因为转账这种高频短事务场景频繁建连开销大。autocommitFalse是关键保证事务由我们手动控制。4.2 转账核心逻辑# transfer.py import time import logging from decimal import Decimal from db import get_conn logger logging.getLogger(transfer) class InsufficientBalance(Exception): pass class AccountNotFound(Exception): pass class TransferService: def __init__(self, max_retry3, retry_interval0.2): self.max_retry max_retry self.retry_interval retry_interval def transfer(self, src_id, dst_id, amount): amount Decimal(str(amount)) if amount 0: raise ValueError(转账金额必须大于 0) for attempt in range(1, self.max_retry 1): conn get_conn() try: with conn.cursor() as cur: conn.begin() # 按 acctid 升序加锁避免死锁 first, second sorted([src_id, dst_id]) cur.execute( SELECT acctid, money FROM account WHERE acctid IN (%s, %s) ORDER BY acctid FOR UPDATE, (first, second), ) rows {r[acctid]: r for r in cur.fetchall()} if src_id not in rows: raise AccountNotFound(f付款账户 {src_id} 不存在) if dst_id not in rows: raise AccountNotFound(f收款账户 {dst_id} 不存在) if rows[src_id][money] amount: raise InsufficientBalance( f账户 {src_id} 余额不足 f{rows[src_id][money]} {amount} ) cur.execute( UPDATE account SET money money - %s WHERE acctid %s, (amount, src_id), ) if cur.rowcount ! 1: raise RuntimeError(f账户 {src_id} 扣款失败) cur.execute( UPDATE account SET money money %s WHERE acctid %s, (amount, dst_id), ) if cur.rowcount ! 1: raise RuntimeError(f账户 {dst_id} 入账失败) conn.commit() logger.info(转账成功 %s - %s : %s, src_id, dst_id, amount) return True except pymysql.err.OperationalError as e: conn.rollback() # 1213 死锁 / 1205 锁等待超时可重试 if e.args[0] in (1213, 1205) and attempt self.max_retry: logger.warning(死锁或锁超时第 %d 次重试, attempt) time.sleep(self.retry_interval) continue raise except Exception: conn.rollback() raise finally: conn.close()几个关键点值得展开说。第一SELECT ... FOR UPDATE锁住两行其他事务想改这两行就得等天然解决并发扣款。第二按acctid升序加锁是因为两个事务如果加锁顺序相反A 锁 1 再锁 2B 锁 2 再锁 1就会互相等待形成死锁统一顺序后死锁概率大幅降低。第三rowcount ! 1的判断不能省它确认 UPDATE 真的命中了目标行而不是静默失败。4.3 用 TaoToken 辅助排查事务问题写这段代码时我对FOR UPDATE加锁范围和死锁重试的边界拿不准就把报错和代码片段丢给 TaoToken 的模型对话让它解释1213和1205的区别、重试该放在哪一层。模型对话入口在这里模型对话https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentchat调用方式就是标准的 OpenAI 兼容格式把base_url指向https://taotoken.net/apiapi_key填config.toml里那个# ai_helper.py import tomllib from openai import OpenAI with open(config.toml, rb) as f: cfg tomllib.load(f) client OpenAI( base_urlcfg[taotoken][api_base], api_keycfg[taotoken][api_key], ) def ask(prompt): resp client.chat.completions.create( modelcfg[taotoken][model], messages[{role: user, content: prompt}], timeoutcfg[taotoken][timeout], ) return resp.choices[0].message.content if __name__ __main__: print(ask(MySQL 中 1213 和 1205 错误码分别代表什么事务里该如何处理))这样你写转账逻辑时遇到不确定的事务行为直接问一句就能拿到解释不用中断思路去翻文档。5. 验证请求转账成功与余额不足两类动作代码写完必须验证而且要覆盖正常和异常两条路径。下面用命令行跑两类动作。5.1 转账成功python -c from transfer import TransferService svc TransferService() svc.transfer(1, 2, 200.00) 执行后查库SELECT acctid, name, money FROM account;预期结果账户 1 从 1000.00 变成 800.00账户 2 从 500.00 变成 700.00两条记录同时更新说明事务提交成功。5.2 余额不足触发回滚python -c from transfer import TransferService svc TransferService() try: svc.transfer(2, 1, 99999.00) except Exception as e: print(捕获异常:, e) 预期输出类似捕获异常: 账户 2 余额不足700.00 99999.00。再查库两个账户余额不变——因为异常在commit之前抛出rollback把整个事务撤销了。这正是事务要么都做要么都不做的体现。5.3 并发扣款验证开两个终端同时执行从账户 1 各转 600 元# 终端 A python -c from transfer import TransferService; TransferService().transfer(1, 2, 600.00) # 终端 B几乎同时 python -c from transfer import TransferService; TransferService().transfer(1, 2, 600.00)账户 1 只有 800 元两笔 600 元不可能都成功。正确结果是一笔成功、一笔因余额不足失败最终账户 1 余额为 200.00绝不会出现负数。如果出现负数说明锁没生效或隔离级别不对回去检查FOR UPDATE和 InnoDB 引擎。6. 本篇常见错排查报错pymysql.err.OperationalError: (1213, Deadlock found)两个事务加锁顺序不一致导致。检查代码里是否按acctid升序加锁sorted([src_id, dst_id])这行不能省。已经加了还偶发就靠重试兜底。报错(1205, Lock wait timeout exceeded)有事务长时间不提交锁没释放。排查是否有连接忘了commit或rollback或者autocommit被误开。用SHOW ENGINE INNODB STATUS看当前锁等待。转账后数据没变也没报错八成是commit没执行或者连接autocommitTrue但事务被隐式提交后又回滚。确认autocommitFalse且commit()在try块内、rollback()在except块内。rowcount返回 0 但没抛异常UPDATE 条件没匹配到行比如acctid传成了字符串。检查参数类型acctid是BIGINT传int而不是str。金额出现799.9999999这种小数用了FLOAT或DOUBLE存金额。改成DECIMAL(15,2)Python 侧用Decimal而不是float。rollback报InterfaceError: (0, )连接已断开还调rollback。在except里先判断连接是否还活着或者用连接池的ping机制自动重连。排障时如果拿不准错误码含义可以直接问模型对话把完整报错贴进去比翻文档快。接入相关的配置问题查接入文档更稳妥接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc7. 把 AI 编码接进日常统一 Key 的长期用法这套转账系统跑通后你会发现真正花时间的不是写 SQL而是处理各种边界死锁重试策略、隔离级别选择、连接池大小调优。这些恰好是 AI 编码助手能帮上忙的地方。把 TaoToken 的 Key 配进你的编辑器或 CLI 工具写代码时随时问比在浏览器和 IDE 之间来回切效率高得多。如果你主要用 Claude Code 这类工具做长期编码Anthropic 兼容通道可以直接对接ClaudeCodeAnthropichttps://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentclaude-code-anthropic一个 Key 同时供脚本调用和编辑器使用配置只维护一份config.toml换模型时改一行model字段就行。对于要长期维护的项目这种统一入口比每个工具单独配 Key 省心不少。转账系统这类涉及资金一致性的代码多问一句这个事务边界对不对往往能提前避开线上事故。