133 lines
5.3 KiB
Python
133 lines
5.3 KiB
Python
import pyodbc
|
|
import logging
|
|
from typing import Optional, Dict, Any, List
|
|
from contextlib import contextmanager
|
|
from config.settings import settings
|
|
import time
|
|
|
|
|
|
class SQLServerConnection:
|
|
def __init__(self):
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
def _build_connection_string(self, server_type="main") -> str:
|
|
"""Construye la cadena de conexión para SQL Server"""
|
|
if server_type == "backup":
|
|
return (
|
|
f"DRIVER={{{settings.backup_driver}}};"
|
|
f"SERVER={settings.backup_server},{settings.backup_port};"
|
|
f"DATABASE={settings.backup_database};"
|
|
f"UID={settings.backup_user};"
|
|
f"PWD={settings.backup_password};"
|
|
f"TrustServerCertificate=yes;"
|
|
f"Connection Timeout=30;"
|
|
)
|
|
else: # main server (CONTROLDESK)
|
|
return (
|
|
f"DRIVER={{{settings.database_driver}}};"
|
|
f"SERVER={settings.database_server},{settings.database_port};"
|
|
f"DATABASE={settings.database_name};"
|
|
f"UID={settings.database_user};"
|
|
f"PWD={settings.database_password};"
|
|
f"TrustServerCertificate=yes;"
|
|
f"Connection Timeout=30;"
|
|
)
|
|
|
|
@contextmanager
|
|
def get_connection(self, server_type="main"):
|
|
"""Context manager para manejar conexiones con cleanup automático"""
|
|
connection = None
|
|
try:
|
|
server_name = "CONTROLDESK" if server_type == "main" else "BACKUP"
|
|
self.logger.info(f"Estableciendo conexión con servidor {server_name}...")
|
|
connection_string = self._build_connection_string(server_type)
|
|
connection = pyodbc.connect(
|
|
connection_string,
|
|
timeout=30,
|
|
autocommit=True
|
|
)
|
|
yield connection
|
|
except pyodbc.Error as e:
|
|
self.logger.error(f"Error de conexión SQL Server: {str(e)}")
|
|
raise
|
|
except Exception as e:
|
|
self.logger.error(f"Error inesperado en conexión: {str(e)}")
|
|
raise
|
|
finally:
|
|
if connection:
|
|
try:
|
|
connection.close()
|
|
self.logger.debug("Conexión cerrada correctamente")
|
|
except Exception as e:
|
|
self.logger.warning(f"Error al cerrar conexión: {str(e)}")
|
|
|
|
def execute_query(self, query: str, params: Optional[tuple] = None, server_type="main") -> List[Dict[str, Any]]:
|
|
"""Ejecuta una consulta SELECT y retorna los resultados como lista de diccionarios"""
|
|
try:
|
|
with self.get_connection(server_type) as conn:
|
|
cursor = conn.cursor()
|
|
|
|
start_time = time.time()
|
|
if params:
|
|
cursor.execute(query, params)
|
|
else:
|
|
cursor.execute(query)
|
|
|
|
# Obtener nombres de columnas
|
|
columns = [column[0] for column in cursor.description] if cursor.description else []
|
|
|
|
# Obtener filas y convertir a diccionarios
|
|
rows = cursor.fetchall()
|
|
results = []
|
|
for row in rows:
|
|
row_dict = {}
|
|
for i, value in enumerate(row):
|
|
if i < len(columns):
|
|
row_dict[columns[i]] = value
|
|
results.append(row_dict)
|
|
|
|
execution_time = time.time() - start_time
|
|
self.logger.info(f"Consulta ejecutada en {execution_time:.3f}s. Filas: {len(results)}")
|
|
|
|
return results
|
|
|
|
except pyodbc.Error as e:
|
|
self.logger.error(f"Error ejecutando consulta SQL: {str(e)}")
|
|
self.logger.error(f"Query: {query}")
|
|
if params:
|
|
self.logger.error(f"Params: {params}")
|
|
raise
|
|
except Exception as e:
|
|
self.logger.error(f"Error inesperado ejecutando consulta: {str(e)}")
|
|
raise
|
|
|
|
def test_connection(self, server_type="main") -> Dict[str, Any]:
|
|
"""Prueba la conexión y retorna información del servidor"""
|
|
try:
|
|
with self.get_connection(server_type) as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# Obtener información del servidor
|
|
cursor.execute("SELECT @@VERSION as ServerVersion, DB_NAME() as DatabaseName, GETDATE() as CurrentTime")
|
|
result = cursor.fetchone()
|
|
|
|
return {
|
|
"status": "connected",
|
|
"server_version": result.ServerVersion if result else "Unknown",
|
|
"database_name": result.DatabaseName if result else "Unknown",
|
|
"server_time": result.CurrentTime.isoformat() if result and result.CurrentTime else None,
|
|
"connection_successful": True
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error probando conexión: {str(e)}")
|
|
return {
|
|
"status": "error",
|
|
"error": str(e),
|
|
"connection_successful": False
|
|
}
|
|
|
|
|
|
# Instancia global de la conexión
|
|
db_connection = SQLServerConnection()
|