Python｜pymysql 官方文档整理（样例 + API 手册）：把连接参数改到 TaoToken 的实操清单

1. 从本地 MySQL 到统一通道pymysql 连接参数到底改哪几个如果你写过 Python 连 MySQL大概率用过 pymysql。官方文档首页给出的样例很直白pymysql.connect(hostlocalhost, port3306, userroot, password123456, databasetest, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor)然后cursor.execute插一条、查一条控制台打印{id: 1, password: very-secret}。这套流程在单机开发时没问题但一旦项目里出现多个数据源、多个环境、多个同事共用一套配置问题就来了密码散落在代码里、测试库和线上库参数混用、换一台机器就要重新改一遍 host 和 port。我最近在整理 pymysql 官方文档样例时顺手把连接参数做了一次“收口”不再让每个脚本自己写死host/user/password而是统一走一个 Key/API 通道把连接信息集中管理。这样做的直接好处是本地 Python 项目接入时只需要维护一份配置换环境只改环境变量不动业务代码。pymysql 本身是纯 Python 实现的 MySQL 客户端兼容 DB-API 2.0安装只要pip install PyMySQL对 CPython 3.6 和 MySQL 5.6 都支持所以改造成本很低。这篇文章面向的是已经在用 pymysql、但连接参数管理比较乱的本地项目。我会先对照官方文档把 Connection 对象和 Cursor 对象的常用参数、方法梳理成速查表再给出可复制的 connect 参数模板、环境变量写法最后附一条连接自检脚本验证参数生效且查询返回正常。核心检索词就是 pymysql 连接配置、官方文档样例、API 手册速查适合想把手写连接串改成统一通道的 Python 开发者。需要先说明一点pymysql 的connect()参数里host、port、user、password、database、charset、cursorclass是最常用的七个官方原型里还有connect_timeout、read_timeout、write_timeout、autocommit、local_infile等。改造时并不是把所有参数都搬到远端而是把“身份与地址”这部分抽出来SQL 逻辑和游标类型仍然留在代码里。这样既保留了 pymysql 的灵活性又避免了密码硬编码。2. TaoToken 前置把连接身份抽到统一 Key/API 通道在动手改 pymysql 参数之前先把“为什么要抽”讲清楚。官方样例里password123456是明文写在代码里的本地玩没问题但项目一旦进 Git这就是个隐患。更麻烦的是当你有多个项目、多个数据库实例时每个项目都要记一套 host/port/user/password改一次密码就要全局搜索替换。统一 Key/API 通道的思路是把数据库连接的身份信息也就是那组账号密码和地址交给一个集中管理的服务Python 侧只保留一个 Key 和 Base URL通过它去换取真正的连接参数。TaoToken 在这里扮演的就是这个“统一通道”的角色。它的 API 地址是https://taotoken.net/api你可以在控制台里创建 API Keys把不同项目的连接身份分开管理。对于 pymysql 这种场景实际做法是本地代码不再直接写hostlocalhost而是从环境变量读取一个 Base URL 和 Key再由一个轻量封装去解析出真正的 host/port/user/password。这样业务代码里的pymysql.connect()调用形式不变但参数来源变了。具体操作上你需要先到官网https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content注册并进入控制台在 API Keys 页面创建一个 Key。创建时注意选择对应的权限范围本地开发建议单独建一个 Key不要和线上共用。拿到 Key 之后把它写进本地.env文件而不是提交到 Git。控制台地址是https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewriteAPI Keys 页面是https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite。这里要强调一个边界TaoToken 不是数据库本身它管理的是连接身份和通道真正的 MySQL 实例还是你自己的。所以 pymysql 的 SQL 语法、事务行为、游标类型都不受影响你原来写的INSERT INTO、SELECT、connection.commit()照常工作。改造只发生在“连接建立”这一步。如果你只是想验证模型对话能力可以走模型对话页面如果是长期编码或 Agent 场景可以了解 Coding Plan接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite。把 Key 管起来之后下一步才是改 pymysql 的 connect 参数。顺序不能反先有 Key再谈参数模板。否则你只是把明文密码从一个文件搬到另一个文件没有解决根本问题。3. 可复制配置connect 参数模板与环境变量写法这一节给出可以直接抄的配置。先看官方文档里 Connection 对象的原型常用参数我整理成下面这张对照表方便你决定哪些留在代码、哪些抽到环境变量。参数官方默认值建议来源说明hostNone环境变量数据库地址抽到统一通道port0实际 3306环境变量端口默认 3306userNone环境变量登录用户名password环境变量登录密码禁止硬编码databaseNone代码或环境变量当前库可用 select_db 切换charset代码固定建议 utf8mb4cursorclassCursor代码固定需要字典结果用 DictCursorconnect_timeout10代码固定连接超时秒数read_timeoutNone代码固定读超时建议显式设置write_timeoutNone代码固定写超时建议显式设置autocommitFalse代码固定默认不自动提交需手动 commit环境变量文件.env写法如下注意不要提交到版本库# .env 本地开发配置加入 .gitignore TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的本地开发Key PYMYSQL_DB_HOST127.0.0.1 PYMYSQL_DB_PORT3306 PYMYSQL_DB_USERroot PYMYSQL_DB_PASSWORD你的数据库密码 PYMYSQL_DB_NAMEtest读取环境变量并构造连接参数的封装建议单独放一个db_config.py# db_config.py import os from dotenv import load_dotenv load_dotenv() def get_conn_params(): 从环境变量组装 pymysql 连接参数 return { host: os.getenv(PYMYSQL_DB_HOST, 127.0.0.1), port: int(os.getenv(PYMYSQL_DB_PORT, 3306)), user: os.getenv(PYMYSQL_DB_USER, root), password: os.getenv(PYMYSQL_DB_PASSWORD, ), database: os.getenv(PYMYSQL_DB_NAME, test), charset: utf8mb4, connect_timeout: 10, read_timeout: 30, write_timeout: 30, autocommit: False, }业务代码里调用时保持官方样例的结构只把参数来源换掉# demo_query.py import pymysql.cursors from db_config import get_conn_params params get_conn_params() params[cursorclass] pymysql.cursors.DictCursor connection pymysql.connect(**params) try: with connection: with connection.cursor() as cursor: sql SELECT id, password FROM users WHERE email%s cursor.execute(sql, (webmasterpython.org,)) result cursor.fetchone() print(result) finally: connection.close()如果你用的是 Cline MCP 或 Codex 这类工具配置里同样要写全三件套Base URL、Key、Model ID。以 Codex 的auth.json为例结构大致如下注意路径按你本地实际位置调整{ base_url: https://taotoken.net/api, api_key: sk-你的本地开发Key, model: 你的模型ID }Cline MCP 的 settings 片段则是{ mcpServers: { taotoken: { url: https://taotoken.net/api, apiKey: sk-你的本地开发Key, model: 你的模型ID } } }这三件套缺一不可Base URL 决定请求发到哪Key 决定身份Model ID 决定用哪个模型。pymysql 场景下虽然不直接调模型但如果你在同一个项目里既连数据库又调模型把这两套配置放在同一个.env里管理会更清爽。4. 验证请求连接自检脚本与成功结果配置写完不能靠猜要有一条自检脚本确认参数真的生效。下面这个脚本做三件事建立连接、执行一条查询、打印连接状态和结果。它不依赖users表是否存在先用SELECT 1验证通道再尝试查表。# self_check.py import pymysql.cursors from db_config import get_conn_params def check_connection(): params get_conn_params() params[cursorclass] pymysql.cursors.DictCursor conn None try: conn pymysql.connect(**params) print([OK] 连接建立成功) print(f[INFO] 连接是否开启: {conn.open}) with conn.cursor() as cursor: cursor.execute(SELECT 1 AS ping) row cursor.fetchone() print(f[OK] 通道自检返回: {row}) with conn.cursor() as cursor: cursor.execute(SHOW TABLES LIKE users) exists cursor.fetchone() if exists: cursor.execute(SELECT id, password FROM users LIMIT 1) print(f[OK] users 表查询返回: {cursor.fetchone()}) else: print([WARN] users 表不存在跳过样例查询) print(f[INFO] 服务器版本: {conn.get_server_info()}) except pymysql.err.OperationalError as e: print(f[FAIL] 连接或查询失败: {e}) finally: if conn and conn.open: conn.close() print([INFO] 连接已关闭) if __name__ __main__: check_connection()运行python self_check.py正常输出类似[OK] 连接建立成功 [INFO] 连接是否开启: True [OK] 通道自检返回: {ping: 1} [OK] users 表查询返回: {id: 1, password: very-secret} [INFO] 服务器版本: 8.0.35 [INFO] 连接已关闭看到{ping: 1}说明通道通了看到{id: 1, password: very-secret}说明查询逻辑和官方样例一致。如果users表不存在脚本会打印 WARN 而不是直接崩方便你先确认连接层没问题。这里用到的conn.open、conn.get_server_info()都是官方 Connection 对象的常用方法cursor.fetchone()返回字典是因为我们指定了DictCursor。验证通过后你可以把这条脚本挂到 CI 的冒烟测试里每次改配置跑一遍。注意read_timeout和write_timeout我设成了 30 秒比官方默认的“无超时”更安全避免慢查询把进程挂死。如果你查的是大结果集考虑换成SSCursor无缓冲游标官方文档里说明它适合返回大量数据或慢网络场景代价是不支持反向滚动、只能遍历确定行数。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth改配置的过程中报错基本集中在几类。下面按真实报错对照排查每条都给出定位思路。第一类pymysql.err.OperationalError: (1045, Access denied for user ...)。这通常是user或password没从环境变量正确读到比如.env没加载、变量名拼错、或者 Key 过期。先打印get_conn_params()的返回值确认password不是空字符串。如果用的是统一通道检查 API Key 是否在控制台被禁用。第二类401 Unauthorized。这个报错在调 API 时出现说明 Key 无效或没带上。检查请求头里是否带了Authorization: Bearer sk-...以及 Base URL 是否是https://taotoken.net/api而不是首页地址。注意 API 地址不带 UTM 参数写错会导致路由不到。第三类local proxy failed。这个一般出现在本地网络层说明请求没发出去。先确认本机能不能访问https://taotoken.net/api再检查是否有本地端口占用或防火墙拦截。不要用任何非官方的网络工具去绕直接排查本地网络配置即可。第四类Error reading choices或类似解析错误。这通常发生在模型返回体解析时说明返回的不是预期 JSON。检查 Model ID 是否写对以及请求的 endpoint 是否匹配。pymysql 场景下如果混用了模型调用确认两套配置没有互相覆盖。第五类OAuth相关报错。如果你用的是 Claude Code 或类似工具OAuth 流程没走完会导致鉴权失败。重新走一遍授权确认回调地址和 Key 一致。Claude Code 的接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite里面有完整的配置步骤。排查时记住一个原则先分层再定位。连接层报错看 host/port/user/password鉴权层报错看 Key 和 Base URL解析层报错看 Model ID 和返回体。不要一上来就改代码先用自检脚本把连接层单独跑通。6. 语义一致 CTA把 Key 管起来再谈查询优化pymysql 的官方文档样例本身不复杂复杂的是参数管理。把连接身份抽到统一通道之后你的业务代码可以保持官方样例的写法with connection:、cursor.execute、connection.commit()这些都不用动。真正需要你花时间的是确认 Key 的权限边界、环境变量的加载顺序、以及超时参数的合理值。如果你还没创建 Key先去 API Keys 页面建一个本地开发专用的https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite。接入细节看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite。想先验证模型对话能力走模型对话页面长期编码或 Agent 场景了解 Coding Plan。把 Key 管好之后再回头优化你的 SQL 和游标类型顺序对了后面就顺了。