Python Flask项目中如何持久化存储MySQL连接池对象？

核心实现逻辑 #

把连接池对象绑定到**Flask应用实例（全局应用上下文）**上，应用实例在服务运行期间常驻，所有请求共享同一个实例的属性，完全符合开发规范，也不会随单次请求销毁。

代码修改示例 #

1. 连接池初始化逻辑修改 #

在Flask应用启动时将初始化好的连接池存入应用配置，仅初始化一次：

from flask import Flask
import mysql.connector

def create_db_connection(app: Flask):
    print("-----INITIALISING-----")
    db_pool = mysql.connector.pooling.MySQLConnectionPool(
        pool_name = "BestLiar_Public_API",
        pool_size = 10,
        autocommit = True,
        pool_reset_session = True,
        user = 'user',
        password = 'pass',
        host = 'host',
        database = 'db'
    )
    print("-----DB POOL INITIALISED-----")
    # 存入应用配置，全局持久化存储
    app.config["DB_POOL"] = db_pool

# 应用工厂示例（如果未用工厂模式，直接在实例化app后调用create_db_connection即可）
def create_app():
    app = Flask(__name__)
    # 应用启动阶段执行一次连接池初始化
    create_db_connection(app)
    # 注册路由、中间件等其他逻辑
    return app

2. 连接获取逻辑修改 #

通过current_app（应用上下文代理）获取全局存储的连接池：

from flask import current_app
import mysql.connector

# 自定义异常类需要提前定义
class NoConnectionError(Exception):
    pass
class SystemOverload(Exception):
    pass

class DBSession:
    def __enter__(self):
        try:
            # 从全局应用配置获取连接池对象
            db_pool = current_app.config["DB_POOL"]
            self.con = db_pool.get_connection()
            self.cur = self.con.cursor(dictionary=True)
            if self.con.is_connected():
                return {'cur': self.cur, 'con': self.con}
            else:
                raise NoConnectionError("No database connection", "Pool connection not connected")
        except mysql.connector.PoolError:
            raise SystemOverload("Too many requests, could not process","No pool connection available")
        except:
            raise NoConnectionError("No database connection", "Unknown reason")

    def __exit__(self, type, value, traceback):
        if self.con: 
            self.cur.close()
            # 池化连接调用close()不会销毁物理连接，仅标记为可复用归还到连接池，符合需求
            self.con.close()

3. 路由使用示例 #

@app.route('/demo')
def demo():
    with DBSession() as db:
        db['cur'].execute("SELECT VERSION() as mysql_version")
        result = db['cur'].fetchone()
    return result

注意事项 #

• 之前存到Flask全局对象g不生效的原因是g是请求级上下文，每个请求结束后就会被销毁，而current_app绑定的是应用实例，服务运行期间一直存在。
• 如果用多worker部署（如gunicorn开启多个进程），每个worker会独立初始化一个连接池，总连接数为pool_size * worker数量，需要注意不要超过MySQL的max_connections上限。
• mysql-connector自带的连接池是线程安全的，多线程模式下请求复用连接池不会出现并发问题。

内容的提问来源于stack exchange，提问作者Björn