#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ DATABASE CONNECTIONS - Gestión de Conexiones de Base de Datos ============================================================ Sistema robusto de conexiones con pool, reconexión automática y logging detallado. Modified: 2025-07-30 - Credentials moved to centralized config """ import pymysql import logging import time import os from contextlib import contextmanager from threading import Lock import threading logger = logging.getLogger(__name__) class DatabaseConnectionPool: """ Pool de conexiones de base de datos thread-safe """ def __init__(self, host='localhost', user='asteriskuser', password=None, database='asterisk', max_connections=10): # Load configuration from centralized file config_file = '/etc/centralita-tarot/config.env' config = {} try: if os.path.exists(config_file): with open(config_file, 'r') as f: for line in f: line = line.strip() if line and not line.startswith('#') and '=' in line: key, value = line.split('=', 1) config[key.strip()] = value.strip() # Override defaults with config values self.host = config.get('DB_HOST', host) self.user = config.get('DB_USER', user) self.password = config.get('DB_PASSWORD', password) self.database = config.get('DB_NAME_ASTERISK', database) logger.info(f"Configuration loaded from {config_file}") else: logger.error(f"Configuration file not found: {config_file}") # Fall back to defaults (for backward compatibility during transition) self.host = host self.user = user self.password = password or 'P4nd0r01$' # Temporary fallback self.database = database logger.warning("Using fallback configuration - UPDATE REQUIRED") except Exception as e: logger.error(f"Error loading configuration: {str(e)}") # Fall back to defaults self.host = host self.user = user self.password = password or 'P4nd0r01$' # Temporary fallback self.database = database self.max_connections = max_connections self.connections = [] self.lock = Lock() logger.info(f"Inicializando pool de conexiones DB: {self.user}@{self.host}/{self.database}") logger.info(f"Máximo de conexiones: {max_connections}") def _create_connection(self): """Crea una nueva conexión a la base de datos""" try: conn = pymysql.connect( host=self.host, user=self.user, password=self.password, database=self.database, charset='utf8mb4', autocommit=True, connect_timeout=10, read_timeout=30, write_timeout=30 ) logger.debug(f"Nueva conexión DB creada: thread-{threading.get_ident()}") return conn except Exception as e: logger.error(f"Error creando conexión DB: {str(e)}") raise def get_connection(self): """Obtiene una conexión del pool""" with self.lock: # Intentar reutilizar conexión existente while self.connections: conn = self.connections.pop() try: # Verificar que la conexión sigue activa conn.ping(reconnect=True) logger.debug(f"Conexión reutilizada del pool: thread-{threading.get_ident()}") return conn except Exception as e: logger.warning(f"Conexión del pool inválida, creando nueva: {str(e)}") continue # Crear nueva conexión si no hay disponibles return self._create_connection() def return_connection(self, conn): """Devuelve una conexión al pool""" with self.lock: if len(self.connections) < self.max_connections: try: # Verificar que la conexión sigue válida conn.ping(reconnect=True) self.connections.append(conn) logger.debug(f"Conexión devuelta al pool: thread-{threading.get_ident()}") except Exception as e: logger.warning(f"Conexión inválida, no devuelta al pool: {str(e)}") try: conn.close() except: pass else: # Pool lleno, cerrar conexión try: conn.close() logger.debug(f"Pool lleno, conexión cerrada: thread-{threading.get_ident()}") except: pass @contextmanager def connection(self): """Context manager para conexiones automáticas""" conn = None try: conn = self.get_connection() yield conn finally: if conn: self.return_connection(conn) # Pool global de conexiones db_pool = DatabaseConnectionPool() @contextmanager def get_db_connection(): """ Context manager para obtener conexión de base de datos Uso: with get_db_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT ...") """ with db_pool.connection() as conn: yield conn def execute_query(sql, params=None, fetch_one=False, fetch_all=False): """ Ejecuta una query con manejo de errores y logging Args: sql: Query SQL params: Parámetros para la query fetch_one: Si debe hacer fetchone() fetch_all: Si debe hacer fetchall() Returns: Resultado de la query o None en caso de error """ start_time = time.time() try: with get_db_connection() as conn: cursor = conn.cursor() if params: logger.debug(f"Ejecutando SQL: {sql} con parámetros: {params}") cursor.execute(sql, params) else: logger.debug(f"Ejecutando SQL: {sql}") cursor.execute(sql) result = None if fetch_one: result = cursor.fetchone() elif fetch_all: result = cursor.fetchall() cursor.close() execution_time = time.time() - start_time logger.debug(f"Query ejecutada en {execution_time:.3f}s") if result: logger.debug(f"Resultado: {result}") return result except Exception as e: execution_time = time.time() - start_time logger.error(f"Error ejecutando query ({execution_time:.3f}s): {str(e)}") logger.error(f"SQL: {sql}") if params: logger.error(f"Parámetros: {params}") return None def execute_insert(sql, params=None): """ Ejecuta un INSERT y retorna el ID insertado """ start_time = time.time() try: with get_db_connection() as conn: cursor = conn.cursor() if params: logger.debug(f"Ejecutando INSERT: {sql} con parámetros: {params}") cursor.execute(sql, params) else: logger.debug(f"Ejecutando INSERT: {sql}") cursor.execute(sql) inserted_id = cursor.lastrowid affected_rows = cursor.rowcount cursor.close() execution_time = time.time() - start_time logger.info(f"INSERT ejecutado en {execution_time:.3f}s - ID: {inserted_id}, Filas: {affected_rows}") return inserted_id except Exception as e: execution_time = time.time() - start_time logger.error(f"Error ejecutando INSERT ({execution_time:.3f}s): {str(e)}") logger.error(f"SQL: {sql}") if params: logger.error(f"Parámetros: {params}") return None def execute_update(sql, params=None): """ Ejecuta un UPDATE y retorna el número de filas afectadas """ start_time = time.time() try: with get_db_connection() as conn: cursor = conn.cursor() if params: logger.debug(f"Ejecutando UPDATE: {sql} con parámetros: {params}") cursor.execute(sql, params) else: logger.debug(f"Ejecutando UPDATE: {sql}") cursor.execute(sql) affected_rows = cursor.rowcount cursor.close() execution_time = time.time() - start_time logger.info(f"UPDATE ejecutado en {execution_time:.3f}s - Filas afectadas: {affected_rows}") return affected_rows except Exception as e: execution_time = time.time() - start_time logger.error(f"Error ejecutando UPDATE ({execution_time:.3f}s): {str(e)}") logger.error(f"SQL: {sql}") if params: logger.error(f"Parámetros: {params}") return 0