How to use MySQL database connection pooling across multiple Python modules [duplicate]

Question

I am new to Flask trying to implement MySQL connection pooling without using SQLAlchemy. I did some research and found out how to define a connection pool using mysql-connector-python (8.0.27). The following article from the official MySQL website provides an example.

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()

The code appears pretty straight-forward and tests fine fine when used in a simple Python file. But if I want to use the same connection pool across multiple Python modules, how and where do I define the connection pool so that it can be reused? My app uses a Flask factory function to initialize. Can someone please assist?

At present, my Flask application uses the following database model (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)

I can figure out that I have to modify my get_db() function as follows:

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

My question is: Where and how do I define the connection pool itself as the get_db() procedure may be called from different Python modules or Flask blueprints?

Shoumik Das · Accepted Answer · 2022-01-07 11:59:41Z

1

I figured this out myself. The solution is to create a singleton class to define the connection pool and then instantiate it only once inside the factory function. Once the connection pool is initialised, we can store it in the app.config dictionary object so that it is accessible by other modules throughout the lifespan of the application. We may need to use the current_app reference while creating a connection from the pool.

answered Jan 7, 2022 at 11:59

Shoumik Das

5215 silver badges20 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

How to use MySQL database connection pooling across multiple Python modules [duplicate]

1 Answer 1

Comments

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Linked

Related