Improved connection pool init

src/server/BreCal/local_db.py

@@ -1,58 +1,83 @@
 import mysql.connector
+from mysql.connector import pooling
 import pydapper
 import logging
 import json
 import os
 import sys
 from BreCal.schemas import defs
+
 config_path = None
+_connection_pool = None
 
-def initPool(instancePath, connection_filename="connection_data_devel.json"):
+
+def _load_json(path):
+    with open(path, encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+def _build_pool_config(connection_data, pool_name, pool_size):
+    pool_config = dict(connection_data)
+    pool_config.setdefault("pool_name", pool_name)
+    pool_config.setdefault("pool_size", pool_size)
+    return pool_config
+
+
+def initPool(instancePath, connection_filename="connection_data_devel.json",
+             pool_name="brecal_pool", pool_size=10):
+    """
+    Initialize the MySQL connection pool and load email credentials.
+    """
+    global config_path, _connection_pool
     try:
-        global config_path
+        if config_path is None:
-        if(config_path == None):
+            config_path = os.path.join(instancePath, f'../../../secure/{connection_filename}')
-            config_path = os.path.join(instancePath,f'../../../secure/{connection_filename}') #connection_data_devel.json');
+            
+        config_path = 'C:\\temp\\connection_data_test.json'
 
-        # config_path = "E:/temp/connection_data.json"
         print(config_path)
-
         if not os.path.exists(config_path):
             print('cannot find ' + os.path.abspath(config_path))
             print("instance path", instancePath)
             sys.exit(1)
 
-        f = open(config_path);
+        connection_data = _load_json(config_path)
-        connection_data = json.load(f)
+        if _connection_pool is None:
-        f.close()
+            pool_config = _build_pool_config(connection_data, pool_name, pool_size)
-
+            _connection_pool = pooling.MySQLConnectionPool(**pool_config)
-        conn_from_pool = mysql.connector.connect(**connection_data)
 
+        conn_from_pool = _connection_pool.get_connection()
+        try:
             commands = pydapper.using(conn_from_pool)
-        data = commands.query("SELECT id from `user`")
+            commands.query("SELECT id from `user` LIMIT 1")
             print("DB connection successful")
+        finally:
             conn_from_pool.close()
 
         credentials_file = "email_credentials_devel.json"
         credentials_path = os.path.join(instancePath, f'../../../secure/{credentials_file}')
 
-        # credentials_path = "E:/temp/email_credentials_devel.json"
+        credentials_path = 'C:\\temp\\email_credentials_test.json'
 
         if not os.path.exists(credentials_path):
             print('cannot find ' + os.path.abspath(credentials_path))
             sys.exit(1)
 
-        f = open(credentials_path);
+        defs.email_credentials = _load_json(credentials_path)
-        defs.email_credentials = json.load(f)
-        f.close()
 
     except mysql.connector.PoolError as e:
         logging.error(f"Failed to create connection pool: {e}")
         print(e)
     except Exception as e:
+        logging.error("Failed to initialize DB pool: %s", e)
         print(e)
 
+
 def getPoolConnection():
-    global config_path
+    if _connection_pool is None:
-    f = open(config_path);
+        raise RuntimeError("Connection pool not initialized. Call initPool first.")
-    connection_data = json.load(f)
+    try:
-    return mysql.connector.connect(**connection_data)
+        return _connection_pool.get_connection()
+    except mysql.connector.PoolError as exc:
+        logging.error("Connection pool exhausted: %s", exc)
+        raise