如何跨多个 Python 模块使用 MySQL 数据库连接池

Question

我是 Flask 的新手，尝试使用 SQLAlchemy 实现 MySQL 连接池而。我做了一些研究，发现了如何使用 mysql-connector-python (8.0.27) 定义连接池。下面article官网MySQL提供了一个例子

dbconfig = {
  "database": "test",
  "user":     "joe"
}

cnxpool = mysql.connector.pooling.MySQLConnectionPool(pool_name = "mypool",
                                                      pool_size = 3,
                                                      **dbconfig)
cnx1 = cnxpool.get_connection()
cnx2 = cnxpool.get_connection()

当在简单的 Python 文件中使用时，代码看起来非常简单并且测试得很好。但是，如果我想在多个 Python 模块中使用相同的连接池，我应该如何以及 在哪里 定义连接池以便它可以被重用？我的应用程序使用 Flask 工厂函数进行初始化。有人可以帮忙吗？

目前，我的Flask应用程序使用以下数据库模型（Flask Tutorial 2.0）：

db.py

import mysql.connector
import click
from flask import current_app, g
from flask.cli import with_appcontext

# Function to create a database connection.
def get_db():
    if 'db' not in g:
        g.db=mysql.connector.connect(host="127.0.0.1",user="dbuser",password="test123",database="testdb")
    return g.db

# Function to close an existing database connection.
def close_db(e=None):
    db=g.pop('db',None)
    if db is not None:
        db.close()

# Function to initialize the database from a script.
def init_db():
    db = get_db()
    # current_app is necessary to specify the location of the file relative to the application directory.
    with current_app.open_resource('schema.sql') as f:
        scriptline=f.readline()
        # Checks for end of script file.
        # Ensure that schema.sql stores each command in a single line.
        while scriptline:
            cursor=db.cursor()
            cursor.execute(scriptline)
            cursor.close()
            scriptline=f.readline()
    db.commit()

@click.command('init-db')
@with_appcontext
def init_db_command():
    # Delete existing data and re-create tables.
    init_db()
    click.echo('Initialized the test database.')

# Register with the application.
def init_app(app):
    app.teardown_appcontext(close_db)
    app.cli.add_command(init_db_command)

我知道我必须按如下方式修改我的 get_db() 函数：

# Function to create a database connection.
def get_db():
    if 'db' not in g:
        g.db=cnxpool.get_connection()
    return g.db

我的问题是：我在哪里以及如何定义连接池本身，因为可以从不同的 Python 模块或 Flask 蓝图调用 get_db() 过程？

Answer 1

这是我自己想出来的。解决方案是创建一个单例class来定义连接池，然后在工厂函数内部只实例化一次。连接池初始化后，我们可以将其存储在 app.config 字典对象中，以便其他模块在应用程序的整个生命周期中都可以访问它。我们可能需要在从池中创建连接时使用 current_app 引用。

如何跨多个 Python 模块使用 MySQL 数据库连接池

How to use MySQL database connection pooling across multiple Python modules

python

mysql

flask