1. 为什么你的 PyMySQL 连接总是断从一次线上事故说起PyMySQL 是用纯 Python 写的 MySQL 客户端库它把「连接」这件事抽象成一个Connection对象你所有的查询、事务、游标都挂在这个对象上。适合谁适合所有用 Python 直接操作 MySQL 的后端开发、数据脚本作者、定时任务维护者。它能做什么一句话让你用 Python 代码完成建连、执行 SQL、提交事务、回滚、关闭这一整套生命周期动作。我见过太多项目把pymysql.connect()写成一个全局变量然后跑着跑着就报(2006, MySQL server has gone away)或者(2013, Lost connection to MySQL server during query)。原因往往不是 MySQL 挂了而是Connection对象的生命周期没管好连接空闲超过wait_timeout被服务端单方面掐断客户端却还以为连接活着下一次查询直接炸。这篇就围绕Connection对象本身来讲。不是泛泛介绍参数表而是把「建连 → 游标 → 事务 → 复用 → 排障」这条链路拆开给你能直接复制的代码。同时我会演示一个容易被忽略的点多环境开发/测试/生产的数据库凭证怎么统一管理避免把密码硬编码在connect()里。这里我用 TaoToken 的统一 Key/API 通道来托管凭证官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 后面第 2 节会讲清楚它和数据库连接的关系。先明确一个认知Connection不是「一次查询的工具」而是一个有状态的会话。它维护着 socket、事务上下文、字符集、当前数据库、autocommit 标志。你把它当短连接用完就关或者当长连接一直不 ping都会出问题。理解它的状态机才是稳健用法的起点。2. TaoToken 前置用统一通道托管多环境数据库凭证在讲Connection参数之前先解决一个现实问题你的connect()里那串passwordxxx从哪来很多人的做法是写死在代码里或者塞进.env然后提交到仓库。前者改密码要改代码后者是安全事故的常客。我的做法是把数据库凭证这类敏感配置通过 TaoToken 的统一 Key/API 通道来管理。TaoToken 提供的是一个统一的 API 入口你可以把它理解成「凭证和模型/服务访问的统一网关」。它的 API 地址是 https://taotoken.net/api 注意这个地址不带任何查询参数是干净的接口根路径。具体怎么和 PyMySQL 配合思路是这样的数据库的 host、user、password 这些不直接写在 Python 文件里而是通过环境变量注入环境变量的值来自你在 TaoToken 控制台配置的凭证。这样开发、测试、生产三套环境用同一份代码只切换环境变量即可。你需要先在 TaoToken 控制台创建 API Key控制台入口在这里https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。创建好 Key 之后在 API Keys 页面可以管理你的密钥https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。这里要澄清一个常见误解TaoToken 不是数据库代理它不会替你转发 MySQL 的 TCP 流量。它管的是「凭证的分发与访问控制」。你的 PyMySQL 依然直连你的 MySQL 实例只是连接参数从统一通道取而不是散落在代码各处。这样做的收益是轮换密码时只改一处审计时知道谁在什么时候取了哪套凭证。如果你还想在写代码时随时验证模型或调试 SQL 生成逻辑可以用模型对话入口https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。长期做编码和 Agent 任务的话Coding Plan 更合适https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。把凭证管理前置讲清楚是因为后面所有Connection初始化代码都会引用这些环境变量。你先把这套通道搭好再往下看配置代码就能直接跑。3. 可复制的 Connection 初始化配置参数、游标与事务现在进入正题。pymysql.connect()返回的就是一个Connection对象。它的构造参数很多但生产环境真正需要你显式设置的其实就那么几个。我先把一份可直接复制的配置给你再逐项解释。import os import pymysql from pymysql.cursors import DictCursor def get_connection(): conn pymysql.connect( hostos.environ[DB_HOST], portint(os.environ.get(DB_PORT, 3306)), useros.environ[DB_USER], passwordos.environ[DB_PASSWORD], databaseos.environ[DB_NAME], charsetutf8mb4, cursorclassDictCursor, autocommitFalse, connect_timeout10, read_timeout30, write_timeout30, max_allowed_packet16 * 1024 * 1024, ) return conn这份配置里有几个关键决策我逐个说。charsetutf8mb4必须显式写。默认的charset会让 PyMySQL 用服务端默认字符集很多老 MySQL 默认是latin1存中文直接乱码。utf8mb4才能存 emoji 和四字节字符。cursorclassDictCursor让查询结果以字典返回而不是元组。这样你写row[user_name]而不是row[3]字段顺序变了也不怕。代价是内存略高但可读性收益远大于此。autocommitFalse是重点。PyMySQL 默认就是False意味着你必须显式commit()或rollback()。很多人以为执行完execute()数据就进库了其实没有连接一关就回滚。这个默认值是对的它逼你思考事务边界。connect_timeout10是建连超时取值 1 到 31536000。read_timeout和write_timeout是读写超时默认None表示永不超时——这在生产环境是危险的一个慢查询能把你的线程挂死。设成 30 秒是合理起点。max_allowed_packet默认 16MB主要影响LOAD DATA LOCAL INFILE这类大批量操作。如果你要导入大文件调大它。关于凭证前面说的环境变量在这里体现为os.environ[DB_HOST]等。你可以写一个.env文件配合python-dotenv加载但.env本身不要提交到仓库。更规范的做法是通过 TaoToken 控制台下发本地开发时用 CLI 拉取到环境变量。如果你用配置文件而不是环境变量可以写一份 TOML[mysql] host 127.0.0.1 port 3306 user app_user database app_db charset utf8mb4 connect_timeout 10 read_timeout 30 write_timeout 30 autocommit false然后读取import tomllib with open(config.toml, rb) as f: cfg tomllib.load(f)[mysql] conn pymysql.connect( hostcfg[host], portcfg[port], usercfg[user], passwordos.environ[DB_PASSWORD], databasecfg[database], charsetcfg[charset], connect_timeoutcfg[connect_timeout], read_timeoutcfg[read_timeout], write_timeoutcfg[write_timeout], autocommitcfg[autocommit], )注意密码依然走环境变量配置文件里不放密码。这是「配置与密钥分离」的基本原则。游标创建用conn.cursor()不传参数就是默认Cursor传DictCursor就是字典游标。你也可以在connect()里用cursorclass设全局默认这样每次cursor()都自动是字典。事务方面conn.begin()显式开启事务conn.commit()提交conn.rollback()回滚。autocommitFalse时第一条 SQL 执行会自动开启事务但显式begin()更清晰。4. 验证请求与成功结果一次完整的事务提交演示配置写好了得验证它真的能跑通。下面这段代码演示一个完整流程建连、建表、插入、提交、查询、关闭。你可以直接复制运行。import pymysql from pymysql.cursors import DictCursor conn pymysql.connect( host127.0.0.1, port3306, userroot, passwordyour_password, databasetest_db, charsetutf8mb4, cursorclassDictCursor, autocommitFalse, ) try: with conn.cursor() as cursor: cursor.execute( CREATE TABLE IF NOT EXISTS users ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(64) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 ) cursor.execute( INSERT INTO users (name) VALUES (%s), (alice,) ) conn.commit() print(insert committed, lastrowid , cursor.lastrowid) with conn.cursor() as cursor: cursor.execute(SELECT id, name, created_at FROM users WHERE name%s, (alice,)) row cursor.fetchone() print(query result:, row) finally: conn.close()跑通后你会看到类似输出insert committed, lastrowid 1 query result: {id: 1, name: alice, created_at: datetime.datetime(2024, 5, 20, 10, 30, 0)}这里有几个细节值得说。with conn.cursor() as cursor会在块结束时自动关闭游标但不会提交事务。所以conn.commit()必须显式调用且要在游标块之外或之内都行只要在close()之前。cursor.lastrowid是刚插入行的自增 ID只在 INSERT 后有效。cursor.rowcount是受影响行数UPDATE 和 DELETE 后常用。验证回滚也很重要。把conn.commit()换成conn.rollback()再查一次你会发现数据没进去。这就是事务的意义。如果你想验证连接是否还活着用conn.ping(reconnectTrue)。它会检查服务器可用性如果连接已断且reconnectTrue会自动重连。注意重连后你之前的事务上下文会丢失所以 ping 一般用在「取连接时」而不是「事务中间」。conn.open属性返回布尔值表示连接是否打开。conn.select_db(other_db)可以切换当前数据库不用重连。一个完整的成功验证应该覆盖建连成功、DDL 执行、DML 执行、commit 生效、查询可见、rollback 生效、close 无异常。这七步都过了你的Connection用法才算基本正确。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或docker ps。再确认端口对默认 3306如果你改了端口port参数要跟着改。如果 MySQL 在容器里host不能写127.0.0.1要写容器网络里的服务名或宿主 IP。报错二pymysql.err.OperationalError: (1045, Access denied for user app_userlocalhost)这是认证失败等价于 HTTP 里的 401。密码错了或者用户没有从该 host 连接的权限。MySQL 的权限是userhost粒度app_userlocalhost和app_user%是两回事。检查SELECT user, host FROM mysql.user;。如果你用 TaoToken 管理凭证确认拉取的是当前环境的那套别把测试环境的密码用到生产。报错三pymysql.err.OperationalError: (2006, MySQL server has gone away)连接被服务端掐了。最常见原因是空闲时间超过wait_timeout默认 28800 秒8 小时。解决方案有两个一是用连接池取连接时ping(reconnectTrue)二是缩短连接生命周期用完就关。长连接场景必须配 ping。报错四pymysql.err.InterfaceError: (0, )这个错误信息很空通常是连接已关闭还在用。检查是不是在conn.close()之后又执行了 SQL或者多线程共享了同一个Connection。PyMySQL 的Connection不是线程安全的每个线程要有自己的连接。报错五local proxy failed或reading choices类错误这类错误通常出现在你通过某个中间层访问服务时。如果你在用统一通道管理凭证确认 API 地址写的是 https://taotoken.net/api 不要多加路径或参数。reading choices往往意味着返回体不是预期的 JSON可能是网络中断或认证头缺失。检查你的请求头里 Key 是否正确带上。报错六OAuth 相关错误如果你用 OAuth 方式获取访问令牌报错通常是 token 过期或 scope 不足。重新走一遍授权流程确认 scope 包含了你需要的权限。令牌要存在环境变量里不要硬编码。报错七RuntimeError: cryptography is required for sha256_passwordMySQL 8 默认用caching_sha2_password认证插件PyMySQL 需要cryptography包。pip install cryptography即可。或者把用户改成mysql_native_password但不推荐安全性差。排查的通用思路先看错误码2003 是网络1045 是认证2006 是超时2013 是查询中断。错误码定位了方向再去查对应配置。6. 语义一致的收尾把 Connection 当成有状态会话去管理写到这里核心的东西都覆盖了。我想再强调一个观念Connection对象不是无状态的函数调用它是有状态的会话。它的状态包括 socket 连接、事务上下文、字符集、当前数据库、autocommit 标志。你每一次execute()都在这个会话里留下痕迹。所以稳健用法的本质是明确它的创建时机、使用边界、销毁时机。短任务用短连接用完即关长任务用连接池取连接时 ping事务要显式 commit 或 rollback多线程不共享连接凭证不硬编码。如果你要把这套用法沉淀成团队规范建议把连接参数抽成配置凭证走统一通道。TaoToken 的接入文档在这里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面有完整的 Key 管理和 API 调用说明。Claude Code 相关的接入配置也可以参考https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。最后给你一个实用技巧在get_connection()里加一行日志记录建连耗时和当前环境。线上出问题时这行日志能帮你快速判断是建连慢还是查询慢。连接泄漏的排查也简单在close()前后打点统计打开和关闭的次数差值持续增长就是泄漏。