企业官网建设流程全解析

1. Flask 里那个让人头大的 MySQL Connection not available

如果你正在用 Flask 写 Web，数据库层用的是mysql-connector-python或者PyMySQL，某天突然在日志里看到这么一行：

mysql.connector.errors.OperationalError: MySQL Connection not available.

然后你会发现，不只是当前这个请求挂了，后面所有涉及数据库的操作全部报同一个错，重启服务才能恢复。这个报错在 Flask + MySQL 的组合里非常典型，尤其是小项目、内部工具、课程作业这类没有专门 DBA 兜底的场景。

它到底是什么意思？简单说，就是你的代码拿着一个「已经不可用」的连接对象去开 cursor 或者执行 SQL。连接不可用的原因有好几类：连接池里的连接被 MySQL 服务端超时踢掉了、连接被某个未读完的结果集搞脏了、配置里连的库和实际用的库错位了。这三类问题的表现都是同一句报错，但排查路径完全不同。

这篇就按「连接池耗尽 / 超时回收 / 配置错位」三个角度拆开讲，给你可以直接复制的 Flask SQLAlchemy 连接池参数、一个健康检查路由、以及用 TaoToken 统一 Key 管理多环境配置的settings.json/config.toml骨架。目标很明确：让你在十分钟内定位到是哪一类问题，并且把连接恢复。

适合谁看：正在用 Flask 写 Web、数据库连接偶发或必现失败、想搞清楚连接池到底怎么配的开发者。不需要你是 DBA，但需要你能改配置文件、能跑一条curl验证。

2. 先搞清楚：连接为什么会被判定为「不可用」

在动手改配置之前，得先理解 MySQL 连接的生命周期。一个 TCP 连接建立后，MySQL 服务端有一个wait_timeout参数，默认 28800 秒（8 小时），空闲超过这个时间，服务端会主动断开。但客户端这边并不知道连接已经断了，它还以为连接是好的，直到下一次拿这个连接去执行 SQL，才会收到「Connection not available」。

另一类更隐蔽：你打开了一个 unbuffered cursor，执行了查询，但没有把结果集读完就close()了。MySQL 协议规定，未读完的结果集会让这个连接处于「脏」状态，后续任何查询都会失败。这就是很多老文章里提到的那个经典 bug，解决方式是在关闭前把剩余结果fetch干净。

第三类是配置错位：你的 Flask 应用连的是 A 库，但迁移脚本或者某个后台任务连的是 B 库，两边连接池参数不一致，导致某一边的连接被大量占用后，另一边拿不到可用连接。

这三类的共同点是：报错信息一样，但修复动作完全不同。所以第一步不是急着改代码，而是先判断你属于哪一类。

2.1 用最小复现确认问题类型

写一个最小的 Flask 路由，故意制造「连接被服务端断开」的场景：

# app.py import time from flask import Flask from sqlalchemy import create_engine, text app = Flask(__name__) # 故意把 pool_recycle 设得比 MySQL wait_timeout 大 engine = create_engine( "mysql+pymysql://user:pass@127.0.0.1:3306/demo", pool_size=5, max_overflow=2, pool_recycle=3600, # 1 小时回收，但服务端可能 10 分钟就断 pool_pre_ping=False, # 关闭预检，模拟「拿到坏连接」 ) @app.route("/bad") def bad(): with engine.connect() as conn: conn.execute(text("SELECT 1")) time.sleep(20) # 模拟空闲 with engine.connect() as conn: conn.execute(text("SELECT 1")) return "ok"

把 MySQL 的wait_timeout临时调小（比如 10 秒），然后访问/bad，你大概率会复现Connection not available。这一步的目的是确认：你的问题是不是「连接被服务端回收但客户端不知道」。

如果复现成功，那pool_pre_ping=True和合理的pool_recycle就是解药。如果复现不出来，那问题更可能在「未读完结果集」或「配置错位」上。

3. TaoToken 前置：把多环境 Key 和连接配置统一管起来

排查连接问题时，最烦的是配置散落在.env、config.py、docker-compose.yml好几个地方，改一个参数要翻三个文件。我习惯把模型调用的 Key 和数据库连接配置都收拢到一份配置骨架里，用 TaoToken 统一管理 Key，这样本地、测试、生产三套环境的差异只体现在一份文件里。

TaoToken 在这里的角色是：你通过它拿到统一的 API Key，用于模型对话、Coding Plan 等场景，同时把数据库连接参数和 Key 放在同一份settings.json或config.toml里，避免「Key 在 A 文件、连接串在 B 文件」的错位。官网入口在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 地址是 https://taotoken.net/api 。

你需要先拿到 Key，再去控制台确认模型和额度。拿 Key 的入口是 API Keys 页面，接入文档在 doc 页面。这两个地址后面 CTA 会再给一次，这里先记住：Key 拿到后不要硬编码进代码，放进配置文件，用环境变量注入。

3.1 settings.json 骨架

{ "taotoken": { "api_key": "${TAOTOKEN_API_KEY}", "base_url": "https://taotoken.net/api", "default_model": "claude-sonnet" }, "database": { "host": "127.0.0.1", "port": 3306, "user": "app_user", "password": "${DB_PASSWORD}", "name": "demo", "charset": "utf8mb4" }, "pool": { "pool_size": 5, "max_overflow": 10, "pool_recycle": 280, "pool_pre_ping": true, "pool_timeout": 30 } }

注意pool_recycle设成 280 秒，比 MySQL 默认wait_timeout小很多，这样连接在被服务端踢掉之前，客户端自己先回收重建。pool_pre_ping打开后，每次从池里取连接会先发一个轻量 ping，坏连接直接丢弃。

3.2 config.toml 骨架

如果你更喜欢 TOML：

[taotoken] api_key = "${TAOTOKEN_API_KEY}" base_url = "https://taotoken.net/api" default_model = "claude-sonnet" [database] host = "127.0.0.1" port = 3306 user = "app_user" password = "${DB_PASSWORD}" name = "demo" charset = "utf8mb4" [pool] pool_size = 5 max_overflow = 10 pool_recycle = 280 pool_pre_ping = true pool_timeout = 30

两份骨架的字段含义一致，选你项目里已经在用的格式就行。关键是：数据库连接参数和 TaoToken 的 Key 放在同一份配置里，改环境时只改这一份。

4. 可复制配置：Flask SQLAlchemy 连接池参数怎么填

拿到配置骨架后，把它接进 Flask。下面是一份可以直接跑的app.py，包含连接池参数、健康检查路由和错误处理。

# app.py import os import json from flask import Flask, jsonify from sqlalchemy import create_engine, text from sqlalchemy.exc import OperationalError app = Flask(__name__) def load_config(path="settings.json"): with open(path, "r", encoding="utf-8") as f: raw = f.read() # 简单替换环境变量占位符 for key, val in os.environ.items(): raw = raw.replace("${" + key + "}", val) return json.loads(raw) cfg = load_config() db = cfg["database"] pool = cfg["pool"] dsn = ( f"mysql+pymysql://{db['user']}:{db['password']}" f"@{db['host']}:{db['port']}/{db['name']}?charset={db['charset']}" ) engine = create_engine( dsn, pool_size=pool["pool_size"], max_overflow=pool["max_overflow"], pool_recycle=pool["pool_recycle"], pool_pre_ping=pool["pool_pre_ping"], pool_timeout=pool["pool_timeout"], future=True, ) @app.route("/health/db") def health_db(): try: with engine.connect() as conn: conn.execute(text("SELECT 1")) return jsonify({"db": "ok"}), 200 except OperationalError as e: return jsonify({"db": "fail", "error": str(e)}), 503 @app.route("/users") def users(): try: with engine.connect() as conn: rows = conn.execute(text("SELECT id, name FROM users LIMIT 10")).fetchall() return jsonify([{"id": r[0], "name": r[1]} for r in rows]) except OperationalError as e: return jsonify({"error": "db_unavailable", "detail": str(e)}), 503

几个参数的解释，用表格对照一下：

pool_recycle是最关键的一个。很多人只改pool_pre_ping，但预检本身有开销，而且如果连接已经被服务端断开，预检只是帮你发现并重建，不如主动回收来得干净。两个一起用最稳。

4.1 未读完结果集的修复

如果你确认问题出在 unbuffered cursor 上，修复方式是在close()之前把剩余结果读干净：

def execute_sql(conn, command, params=None, search=True): cursor = conn.cursor() try: if params is None: cursor.execute(command) else: cursor.execute(command, params) if search: return cursor.fetchall() else: # 关键：把剩余结果读干净，避免连接变脏 while cursor.fetchone() is not None: pass conn.commit() return cursor.rowcount finally: cursor.close()

注意finally里关闭 cursor，保证异常时也能释放。while cursor.fetchone()这个循环看起来笨，但它是解决「未读完结果集导致连接不可用」最直接的办法。

5. 验证请求与成功结果

配置改完后，先跑健康检查：

curl -s http://127.0.0.1:5000/health/db

正常返回：

{"db": "ok"}

如果返回 503，说明连接还是有问题，看error字段里的具体信息。接着验证业务路由：

curl -s http://127.0.0.1:5000/users

正常返回用户列表。然后做一次「空闲后恢复」测试：等 5 分钟（超过你设的pool_recycle），再访问一次/users，如果还能正常返回，说明连接回收和预检生效了。

再做一个并发测试，确认池参数合理：

for i in $(seq 1 20); do curl -s http://127.0.0.1:5000/users & done wait

如果 20 个并发请求都返回正常，没有出现QueuePool limit或Connection not available，说明pool_size + max_overflow够用。如果出现排队超时，把max_overflow调大一点。

5.1 用 TaoToken 验证模型调用是否正常

数据库通了之后，顺手验证一下 TaoToken 的 Key 是否可用。用模型对话页面发一条测试消息，或者用 API 直接调：

curl -s https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "claude-sonnet", "messages": [{"role": "user", "content": "ping"}] }'

返回里有choices字段就说明 Key 正常。这一步和数据库排查是独立的，但放在一起做，能确认你的配置骨架里两类 Key 都没问题。

6. 本篇常见错排查

6.1 报错依旧是 Connection not available，但 pool_pre_ping 已开

检查pool_recycle是不是大于 MySQL 的wait_timeout。用这条 SQL 查服务端超时：

SHOW VARIABLES LIKE 'wait_timeout';

如果返回 28800，而你的pool_recycle是 3600，那连接在池里待了 1 小时才回收，但服务端 8 小时才断，理论上没问题。但如果服务端被改成了 600，而你的pool_recycle是 3600，那连接在池里待到 600 秒后就被服务端断了，客户端却要等到 3600 秒才回收，中间这段时间拿到的都是坏连接。把pool_recycle设成比wait_timeout小 20–30 秒。

6.2 健康检查通过，但业务路由偶发失败

大概率是未读完结果集。检查你的代码里有没有cursor.execute()之后只fetchone()一次就close()的地方。把所有非查询类的execute后面补上「读干净」的逻辑，或者改用 buffered cursor。

6.3 配置错位：连的库和预期不一致

在健康检查路由里加一行，把当前连接的库名打出来：

with engine.connect() as conn: db_name = conn.execute(text("SELECT DATABASE()")).scalar() print("connected to:", db_name)

如果打出来的库名和你以为的不一样，检查settings.json里的database.name，以及环境变量DB_PASSWORD有没有被别的 shell 覆盖。

6.4 TaoToken Key 报 401

先确认TAOTOKEN_API_KEY环境变量在当前 shell 里存在：

echo $TAOTOKEN_API_KEY

如果为空，说明配置文件里的${TAOTOKEN_API_KEY}没被替换。检查load_config里的替换逻辑，或者直接在启动 Flask 前export一下。Key 的管理入口在 API Keys 页面，接入细节看 doc 文档。

7. 把连接池和 Key 都收进一份配置里

回到最开始的问题：MySQL Connection not available之所以难查，是因为它把三类不同的问题包装成了同一句报错。你现在手里有了三样东西：一份带pool_recycle和pool_pre_ping的连接池配置、一个能区分「连接坏了」和「配置错了」的健康检查路由、一份把数据库参数和 TaoToken Key 放在一起的配置骨架。

下一步动作很具体：先跑/health/db确认当前状态，再对照wait_timeout调pool_recycle，最后用并发请求压一下池参数。如果排查过程中需要看模型侧的调用是否正常，模型对话页面可以直接发消息验证；如果是长期编码或 Agent 场景，Coding Plan 页面有对应的套餐说明；Key 的创建和轮换在 API Keys 页面，接入细节在 doc 文档。把这几步走完，连接不可用的问题基本能定位到具体那一类，而不是靠重启服务碰运气。