2024-07-14 16:07:00 +00:00
|
|
|
import mysql.connector
|
2024-07-17 12:01:12 +00:00
|
|
|
from loguru import logger
|
2024-07-14 16:07:00 +00:00
|
|
|
from mysql.connector import pooling
|
2024-03-29 17:17:51 +00:00
|
|
|
|
2024-07-17 12:01:12 +00:00
|
|
|
from lib.constants import CONST
|
|
|
|
|
2023-06-19 14:20:17 +00:00
|
|
|
|
2024-07-14 16:07:00 +00:00
|
|
|
def create_connection_pool(name: str, size: int) -> pooling.MySQLConnectionPool:
|
2024-07-16 19:53:42 +00:00
|
|
|
return pooling.MySQLConnectionPool(
|
2024-07-14 16:07:00 +00:00
|
|
|
pool_name=name,
|
|
|
|
pool_size=size,
|
2024-04-08 19:13:18 +00:00
|
|
|
host="db",
|
|
|
|
port=3306,
|
2024-07-16 19:53:42 +00:00
|
|
|
database=CONST.MARIADB_DATABASE,
|
|
|
|
user=CONST.MARIADB_USER,
|
|
|
|
password=CONST.MARIADB_PASSWORD,
|
2024-07-14 16:07:00 +00:00
|
|
|
charset="utf8mb4",
|
|
|
|
collation="utf8mb4_unicode_ci",
|
2024-04-08 19:13:18 +00:00
|
|
|
)
|
|
|
|
|
2024-04-13 15:38:01 +00:00
|
|
|
|
2024-04-08 19:13:18 +00:00
|
|
|
try:
|
|
|
|
_cnxpool = create_connection_pool("core-pool", 25)
|
2024-07-14 16:07:00 +00:00
|
|
|
except mysql.connector.Error as e:
|
|
|
|
logger.critical(f"Couldn't create the MySQL connection pool: {e}")
|
2024-04-08 19:13:18 +00:00
|
|
|
raise e
|
2023-06-19 14:20:17 +00:00
|
|
|
|
|
|
|
|
|
|
|
def execute_query(query, values=None):
|
2024-07-16 19:53:42 +00:00
|
|
|
with _cnxpool.get_connection() as conn:
|
|
|
|
with conn.cursor() as cursor:
|
|
|
|
cursor.execute(query, values)
|
|
|
|
conn.commit()
|
|
|
|
return cursor
|
2023-06-19 14:20:17 +00:00
|
|
|
|
|
|
|
|
|
|
|
def select_query(query, values=None):
|
2024-07-16 19:53:42 +00:00
|
|
|
with _cnxpool.get_connection() as conn:
|
|
|
|
with conn.cursor() as cursor:
|
|
|
|
cursor.execute(query, values)
|
|
|
|
return cursor.fetchall()
|
2023-06-19 14:20:17 +00:00
|
|
|
|
|
|
|
|
|
|
|
def select_query_one(query, values=None):
|
2024-07-16 19:53:42 +00:00
|
|
|
with _cnxpool.get_connection() as conn:
|
|
|
|
with conn.cursor() as cursor:
|
|
|
|
cursor.execute(query, values)
|
|
|
|
output = cursor.fetchone()
|
|
|
|
return output[0] if output else None
|