Flask+Peewee环境下MySQL连接超时及数据库宕机处理咨询

一、设置数据库连接超时 #

不同数据库的连接超时参数由底层驱动定义，Peewee允许直接将这些参数传入数据库初始化方法：

PostgreSQL示例 #

from peewee import PostgresqlDatabase

db = PostgresqlDatabase(
    "your_db_name",
    user="db_user",
    password="db_pass",
    host="db_host",
    connect_timeout=5  # 设置5秒连接超时
)

MySQL示例 #

from peewee import MySQLDatabase

db = MySQLDatabase(
    "your_db_name",
    user="db_user",
    password="db_pass",
    host="db_host",
    connect_timeout=5  # 同样设置5秒超时
)

注意：SQLite为本地数据库，无网络连接环节，无需设置连接超时。

二、正确处理数据库宕机场景 #

你之前通过connect()捕获异常无效，核心原因是Peewee默认采用懒连接机制——仅在执行查询操作时才会实际建立数据库连接，提前调用connect()无法覆盖后续查询时的连接失败情况。以下是可靠的处理方案：

1. 全局异常捕获（Flask专属） #

利用Flask的错误处理器统一捕获数据库连接相关异常：

from peewee import OperationalError
from flask import jsonify

@app.errorhandler(OperationalError)
def handle_db_unavailable(error):
    return jsonify({"error": "数据库服务暂时不可用，请稍后重试"}), 503

2. 局部查询异常处理 #

在特定视图或业务逻辑中，针对查询操作捕获异常并尝试恢复：

@app.route("/api/data")
def get_data():
    try:
        # 执行实际查询
        records = YourModel.select().where(YourModel.status == 1)
        return jsonify([r.to_dict() for r in records])
    except OperationalError:
        # 先关闭无效连接，尝试重新连接后重试
        db.close()
        try:
            db.connect()
            records = YourModel.select().where(YourModel.status == 1)
            return jsonify([r.to_dict() for r in records])
        except OperationalError:
            return jsonify({"error": "数据库连接失败"}), 503

3. 使用连接池自动管理失效连接 #

启用Peewee的连接池功能，它会自动检测并移除失效连接，减少手动处理的复杂度：

from peewee import PooledPostgresqlDatabase

db = PooledPostgresqlDatabase(
    "your_db_name",
    user="db_user",
    password="db_pass",
    host="db_host",
    connect_timeout=5,
    max_connections=20,  # 最大连接数，根据并发量调整
    stale_timeout=300    # 超过5分钟的连接将被标记为失效并自动清理
)

三、关键注意事项 #

• 不要依赖connect()提前验证连接：懒连接机制下，连接状态可能在查询时发生变化
• 重点捕获OperationalError：这是Peewee在数据库连接失败、超时或宕机时抛出的核心异常类型
• 连接池参数需适配业务场景：max_connections和stale_timeout需根据并发量和数据库稳定性调整

内容的提问来源于stack exchange，提问作者Mickael