DBUtils is a Python library designed to provide connection pooling functionality for database operations.

There are two primary modes of connection handling:

Mode One: Each thread maintains its own dedicated connection. Even if the close method is invoked, the connection is returned to the pool rather than being terminated, allowing reuse within the same thread. When the thread ends, the connection is automatically closed.

import pymysql
from DBUtils.PersistentDB import PersistentDB

POOL = PersistentDB(
    creator=pymysql,
    maxusage=None,
    setsession=[],
    ping=0,
    closeable=False,
    threadlocal=None,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='mysql',
    database='python',
    charset='utf8'
)

def execute_query():
    conn = POOL.connection(shareable=False)
    cursor = conn.cursor()
    cursor.execute('select * from students')
    data = cursor.fetchall()
    print(data)
    cursor.close()
    conn.close()

execute_query()

Running this code produces the following output:

Mode Two: A fixed number of connections are pre-created and shared among all thread. Note: Since pymysql and MySQLdb have threadsafety value of 1, all connections in thiss mode can be shared across threads.

import time
import pymysql
import threading
from DBUtils.PooledDB import PooledDB, SharedDBConnection

POOL = PooledDB(
    creator=pymysql,
    maxconnections=6,
    mincached=2,
    maxcached=5,
    maxshared=3,
    blocking=True,
    maxusage=None,
    setsession=[],
    ping=0,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='mysql',
    database='python',
    charset='utf8'
)

def execute_query():
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute('select * from students')
    data = cursor.fetchall()
    print(data)
    conn.close()

execute_query()

Output appears as follows:

Using DBUtils Connection Pool in Flask Applications

Without Object Instantiation

Create a file named db_helper.py in the project root:

import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,
    maxconnections=6,
    mincached=2,
    maxcached=5,
    maxshared=3,
    blocking=True,
    maxusage=None,
    setsession=[],
    ping=0,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='mysql',
    database='python',
    charset='utf8'
)

class DatabaseManager(object):
    
    @staticmethod
    def get_single_record(query, params):
        connection = POOL.connection()
        cursor = connection.cursor()
        cursor.execute(query, params)
        record = cursor.fetchone()
        connection.close()
        return record

    @staticmethod
    def get_all_records(query, params):
        connection = POOL.connection()
        cursor = connection.cursor()
        cursor.execute(query, params)
        records = cursor.fetchall()
        connection.close()
        return records

Create the main application file run.py:

from flask import Flask
from db_helper import DatabaseManager

app = Flask(__name__)

@app.route("/")
def greet():
    record = DatabaseManager.get_single_record('select * from students', [])
    print(record)
    return "Hello World"

if __name__ == '__main__':
    app.run()

Access the following URL:

http://127.0.0.1:5000/

Backend test results:

With Object Instantiation

Create another db_helper.py in the project root:

import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,
    maxconnections=6,
    mincached=2,
    maxcached=5,
    maxshared=3,
    blocking=True,
    maxusage=None,
    setsession=[],
    ping=0,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='mysql',
    database='python',
    charset='utf8'
)

class DatabaseManager(object):
    def __init__(self):
        self.connection = POOL.connection()
        self.cursor = self.connection.cursor()

    def cleanup(self):
        self.cursor.close()
        self.connection.close()

    def get_single_record(self, query, params):
        self.cursor.execute(query, params)
        record = self.cursor.fetchone()
        self.cleanup()
        return record

    def get_all_records(self, query, params):
        self.cursor.execute(query, params)
        records = self.cursor.fetchall()
        self.cleanup()
        return records

Update the main application file run.py:

from flask import Flask
from db_helper import DatabaseManager

app = Flask(__name__)

@app.route("/")
def greet():
    db_instance = DatabaseManager()
    record = db_instance.get_single_record('select * from students', [])
    print(record)
    return "Hello World"

if __name__ == '__main__':
    app.run()

Visit the URL:

http://127.0.0.1:5000/

Test results display: