1. 从一次连接泄漏说起pymysql 连接池到底该怎么配如果你写过 Python 后端大概率写过这样的代码conn pymysql.connect(...)用完conn.close()。单机脚本没问题但一旦上了 Web 服务每个请求都新建连接数据库的max_connections很快被打满报错Too many connections。这时候你会想到连接池。pymysql 本身不自带连接池社区常用的是DBUtils里的PooledDB或者干脆用 SQLAlchemy 的QueuePool包一层。连接池能做什么简单说就是提前建好一批连接放在池子里请求来了借一个用完还回去而不是反复握手、认证、断开。适合谁适合任何有并发请求、又不想上重型 ORM 的 Python 项目尤其是 Flask/FastAPI 里直接写 SQL 的场景。但连接池只解决了一半问题。另一半是配置管理数据库密码、API Key 这类敏感信息散落在settings.py、.env、Docker 环境变量里多环境开发/测试/生产切换时特别容易出错。这篇就把两件事合起来讲用PooledDB搭一个可复制的 pymysql 连接池骨架再用 TaoToken 的统一 Key/API 通道把多环境密钥集中管起来。TaoToken 在这里扮演的是「统一入口」的角色——你不再到处复制粘贴不同的 Key而是通过一个 API 通道拿到配置代码里只认一个地址。我试过把数据库配置和模型调用配置分开管结果每次换环境都要改两处后来统一到 TaoToken 的 Key 体系下才清爽。下面从环境准备开始一步步给你能直接跑的代码。2. 前置准备TaoToken 统一 Key 与项目依赖在动手写连接池之前先把「密钥从哪来」这件事定下来。传统做法是把DB_PASSWORD写进.env但多环境时你得维护好几份。TaoToken 的思路是提供一个统一的 API 通道你用同一个 Key 去访问不同能力模型对话、编码计划等配置集中在一处。第一步去官网注册并拿到 Key。地址是https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content注册后在控制台创建 API Key。控制台入口https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite。Key 的管理页面在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite建议给不同环境建不同的 Key方便单独吊销。第二步装依赖。连接池用DBUtilsHTTP 请求用requests配置读取用标准库json就够不引入额外负担pip install pymysql DBUtils requests第三步理解 TaoToken 的 API 基地址。所有请求走https://taotoken.net/api注意这个地址不带UTM 参数是纯 API 端点。你在代码里把它作为base_url配合Authorization: Bearer 你的Key头就能调用。文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite接入细节以文档为准。这里要强调一点TaoToken 是合规的 API 聚合通道不是让你绕过什么限制的工具。它的价值在于统一管理——一个 Key 管多环境、多能力减少密钥散落带来的泄露风险。数据库密码这类东西我建议仍然放在本地settings.json但通过 TaoToken 的 Key 来做配置服务的鉴权这样配置下发也有身份校验。3. 可复制的连接池配置骨架与 settings.json先给配置文件。项目根目录建一个settings.json把数据库连接参数和 TaoToken 的接入信息分开写{ env: dev, database: { host: 127.0.0.1, port: 3306, user: app_user, password: your_db_password, database: app_db, charset: utf8mb4 }, taotoken: { base_url: https://taotoken.net/api, api_key: sk-你的Key, timeout: 10 }, pool: { mincached: 2, maxcached: 5, maxconnections: 20, blocking: true, ping: 1 } }字段说明用表格对照一下方便你按自己项目调字段含义建议值mincached启动时预建的空闲连接数2~5maxcached池中最多缓存的空闲连接5~10maxconnections池允许的最大连接总数按 DB 上限的 70% 设blocking池满时是否阻塞等待trueping每次取连接前是否探活1开启ping1很关键。MySQL 默认wait_timeout是 8 小时连接放久了会被服务端单方面断开取出来直接用就报Lost connection。开启 ping 后DBUtils 会在借出前发一个轻量探测失效就重建。接下来是连接池封装。新建db_pool.pyimport json import pymysql from dbutils.pooled_db import PooledDB with open(settings.json, r, encodingutf-8) as f: CFG json.load(f) DB CFG[database] POOL_CFG CFG[pool] _pool PooledDB( creatorpymysql, mincachedPOOL_CFG[mincached], maxcachedPOOL_CFG[maxcached], maxconnectionsPOOL_CFG[maxconnections], blockingPOOL_CFG[blocking], pingPOOL_CFG[ping], hostDB[host], portDB[port], userDB[user], passwordDB[password], databaseDB[database], charsetDB[charset], cursorclasspymysql.cursors.DictCursor, ) def get_conn(): return _pool.connection()注意creatorpymysql这一行DBUtils 会用它来创建真实连接。cursorclass设成DictCursor查询结果直接是字典省去手动映射列名。然后是查询封装重点是用完必须归还。用上下文管理器最稳from contextlib import contextmanager contextmanager def db_cursor(commitFalse): conn get_conn() cursor conn.cursor() try: yield cursor if commit: conn.commit() except Exception: conn.rollback() raise finally: cursor.close() conn.close() # 归还到池不是真关闭这里有个容易踩的坑conn.close()在 PooledDB 里是归还连接不是断开。所以别担心频繁 close 有性能问题反而你不 close 才会把池耗干。4. 用 TaoToken 统一 Key 做配置鉴权与密钥切换连接池搭好了现在解决多环境密钥切换。思路是把「当前用哪套数据库配置」这个决策交给 TaoToken 的 Key 来标识。不同环境用不同 Key代码启动时根据 Key 去拉对应配置。先写一个调用 TaoToken 的客户端import requests TT CFG[taotoken] def fetch_remote_config(env: str) - dict: url f{TT[base_url]}/config headers { Authorization: fBearer {TT[api_key]}, Content-Type: application/json, } payload {env: env} resp requests.post(url, jsonpayload, headersheaders, timeoutTT[timeout]) resp.raise_for_status() return resp.json()这段代码里base_url就是https://taotoken.net/apiapi_key从settings.json读。实际接入时接口路径和字段以官方文档为准我这里给的是骨架结构你照着替换即可。密钥切换的验证动作很简单改settings.json里的env字段从dev改成prod重启服务观察日志里拉到的配置是不是变了。为了不重启也能切可以加一个热加载函数def reload_pool(env: str): global _pool, DB remote fetch_remote_config(env) DB remote[database] _pool PooledDB( creatorpymysql, mincachedPOOL_CFG[mincached], maxcachedPOOL_CFG[maxcached], maxconnectionsPOOL_CFG[maxconnections], blockingPOOL_CFG[blocking], pingPOOL_CFG[ping], hostDB[host], portDB[port], userDB[user], passwordDB[password], databaseDB[database], charsetDB[charset], cursorclasspymysql.cursors.DictCursor, )调用reload_pool(prod)就会用新配置重建池。旧池的连接会在归还后自然回收不用手动清理。如果你还想验证模型能力是否正常可以顺手调一下模型对话接口地址是https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite确认 Key 有权限。长期做编码或 Agent 的话Coding Plan 页面在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite可以按需了解。5. 验证请求连接复用与密钥切换的实测结果光写代码不算数得跑起来看结果。先验证连接复用。写一个测试脚本连续取 100 次连接打印每次连接的线程 ID 和连接对象 IDimport threading from db_pool import get_conn def worker(n): conn get_conn() cur conn.cursor() cur.execute(SELECT CONNECTION_ID() AS cid) row cur.fetchone() print(ftask-{n} thread{threading.get_ident()} db_conn_id{row[cid]}) cur.close() conn.close() for i in range(100): worker(i)跑完你会看到db_conn_id只在少数几个值之间循环说明连接被复用了没有每次都新建。如果每次都是新 ID检查maxcached是不是设成了 0。再验证密钥切换。准备两套配置dev指向本地库prod指向另一台。执行from db_pool import reload_pool, db_cursor reload_pool(prod) with db_cursor() as cur: cur.execute(SELECT DATABASE() AS db) print(cur.fetchone())输出应该显示prod对应的库名。如果还是旧库说明reload_pool里的全局变量没生效——Python 里global声明别漏了。成功的结果长这样连接池稳定复用切换环境后查询落到正确的库整个过程不用改一行业务代码只动settings.json里的env。这就是统一 Key 接入的价值配置和代码解耦。6. 本篇常见错排查报错一Too many connections。池的maxconnections设太大或者你根本没归还连接。检查每个get_conn()是否都有对应的close()用db_cursor上下文管理器能避免大部分泄漏。报错二Lost connection to MySQL server during query。连接空闲太久被服务端断了。把ping设成 1或者调大 MySQL 的wait_timeout。生产环境建议两者都做。报错三pymysql.err.OperationalError: (1045, Access denied)。密钥切换后密码不对。确认 TaoToken 拉回来的配置里password字段正确注意别把dev的 Key 用到prod上。报错四RuntimeError: no connection available。池满了且blockingfalse。要么调大maxconnections要么设blockingtrue让它排队等。报错五TaoToken 请求 401。Authorization头格式不对必须是Bearer Key中间一个空格。Key 前后别带引号或换行。排查顺序建议先看日志里连接 ID 是否复用再看配置拉取是否成功最后看数据库侧SHOW PROCESSLIST确认连接数。三步定位基本能覆盖九成问题。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite遇到接口字段对不上时以文档为准。Key 管理记得定期轮换控制台在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite。