local-llm-server/llm_server/database/conn.py

import pymysql
from dbutils.pooled_db import PooledDB


class DatabaseConnection:
    db_pool: PooledDB = None

    def init_db(self, host, username, password, database):
        self.db_pool = PooledDB(
            creator=pymysql,
            maxconnections=10,
            host=host,
            user=username,
            password=password,
            database=database,
            charset='utf8mb4',
            autocommit=True,
        )

        # Test it.
        conn = self.db_pool.connection()
        del conn

    def connection(self):
        return self.db_pool.connection()


db_pool = DatabaseConnection()