"""Gestión de conexiones y operaciones SQL Server.""" import pyodbc from typing import Optional, Tuple, List import time from dataclasses import dataclass from ..utils.logger import app_logger @dataclass class LogicalFile: """Representa un archivo lógico de una base de datos.""" logical_name: str physical_name: str type: str # 'D' (data) o 'L' (log) class SQLServerManager: """Gestor de operaciones con SQL Server.""" def __init__( self, server: str, use_windows_auth: bool = True, username: Optional[str] = None, password: Optional[str] = None ): """ Inicializa el gestor de SQL Server. Args: server: Nombre o IP del servidor SQL Server use_windows_auth: Usar autenticación de Windows username: Usuario SQL (si no usa Windows Auth) password: Contraseña SQL (si no usa Windows Auth) """ self.server = server self.use_windows_auth = use_windows_auth self.username = username self.password = password def get_connection_string(self, database: Optional[str] = None) -> str: """ Construye la cadena de conexión. Args: database: Base de datos específica (opcional) Returns: Cadena de conexión ODBC """ parts = [ f"DRIVER={{ODBC Driver 17 for SQL Server}}", f"SERVER={self.server}" ] if database: parts.append(f"DATABASE={database}") if self.use_windows_auth: parts.append("Trusted_Connection=yes") else: parts.append(f"UID={self.username}") parts.append(f"PWD={self.password}") return ";".join(parts) def test_connection(self) -> Tuple[bool, Optional[str]]: """ Prueba la conexión a SQL Server. Returns: Tupla (éxito, mensaje_error) """ try: conn_str = self.get_connection_string() conn = pyodbc.connect(conn_str, timeout=10) cursor = conn.cursor() cursor.execute("SELECT @@VERSION") version = cursor.fetchone()[0] conn.close() app_logger.info(f"Conexión SQL Server exitosa: {version[:50]}...") return True, None except Exception as e: error_msg = f"Error conectando a SQL Server: {str(e)}" app_logger.error(error_msg) return False, error_msg def get_filelist_from_backup( self, backup_path: str, timeout_minutes: int = 5 ) -> Tuple[List[LogicalFile], Optional[str]]: """ Obtiene la lista de archivos lógicos de un backup usando RESTORE FILELISTONLY. Args: backup_path: Ruta al archivo .bak timeout_minutes: Timeout en minutos Returns: Tupla (lista de LogicalFile, mensaje_error) """ try: conn_str = self.get_connection_string() conn = pyodbc.connect(conn_str, timeout=timeout_minutes * 60) cursor = conn.cursor() start_time = time.time() app_logger.info(f"Ejecutando RESTORE FILELISTONLY FROM DISK = '{backup_path}'") query = f"RESTORE FILELISTONLY FROM DISK = N'{backup_path}'" cursor.execute(query) elapsed = time.time() - start_time app_logger.info(f"FILELISTONLY completado en {elapsed:.2f}s") logical_files = [] for row in cursor.fetchall(): # Las columnas típicas son: LogicalName, PhysicalName, Type, ... logical_name = row.LogicalName physical_name = row.PhysicalName file_type = row.Type logical_files.append(LogicalFile( logical_name=logical_name, physical_name=physical_name, type=file_type )) app_logger.debug( f"Archivo lógico: {logical_name} ({file_type}) -> {physical_name}" ) conn.close() if not logical_files: return [], "No se encontraron archivos lógicos en el backup" return logical_files, None except Exception as e: error_msg = f"Error obteniendo FILELISTONLY: {str(e)}" app_logger.error(error_msg) return [], error_msg def restore_database( self, db_name: str, backup_path: str, data_folder: str, logical_files: List[LogicalFile], timeout_minutes: int = 60, dry_run: bool = False ) -> Tuple[bool, Optional[str], Optional[str]]: """ Restaura una base de datos desde un backup. Args: db_name: Nombre de la base de datos backup_path: Ruta al archivo .bak data_folder: Carpeta donde se crearán los archivos de datos logical_files: Lista de archivos lógicos del backup timeout_minutes: Timeout en minutos dry_run: Si True, solo simula (no ejecuta RESTORE) Returns: Tupla (éxito, stdout, error) """ try: # Construir las cláusulas MOVE move_clauses = [] for lf in logical_files: if lf.type == 'D': # Data file new_path = f"{data_folder}\\{db_name}.mdf" elif lf.type == 'L': # Log file new_path = f"{data_folder}\\{db_name}_log.ldf" else: # Archivos adicionales (filestream, etc.) continue move_clauses.append(f"MOVE N'{lf.logical_name}' TO N'{new_path}'") if not move_clauses: return False, None, "No se pudieron determinar los archivos de datos y log" # Construir el comando RESTORE restore_cmd = f""" -- Poner la base de datos en modo single user ALTER DATABASE [{db_name}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; -- Restaurar RESTORE DATABASE [{db_name}] FROM DISK = N'{backup_path}' WITH {', '.join(move_clauses)}, REPLACE; -- Volver a modo multi user ALTER DATABASE [{db_name}] SET MULTI_USER; """ app_logger.info(f"Comando RESTORE generado:\n{restore_cmd}") if dry_run: app_logger.info("Modo DRY RUN: No se ejecutará el RESTORE") return True, restore_cmd, None # Ejecutar RESTORE conn_str = self.get_connection_string() conn = pyodbc.connect(conn_str, timeout=timeout_minutes * 60) conn.autocommit = True # Necesario para ALTER DATABASE cursor = conn.cursor() start_time = time.time() app_logger.info(f"Ejecutando RESTORE DATABASE [{db_name}]...") # Ejecutar en múltiples pasos output_lines = [] # 1. Single user try: cursor.execute(f"ALTER DATABASE [{db_name}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE") output_lines.append("Base de datos configurada en modo SINGLE_USER") except Exception as e: app_logger.warning(f"Error configurando SINGLE_USER (puede no existir la DB): {e}") # 2. RESTORE restore_query = f"RESTORE DATABASE [{db_name}] FROM DISK = N'{backup_path}' WITH {', '.join(move_clauses)}, REPLACE" cursor.execute(restore_query) output_lines.append("RESTORE DATABASE completado") # 3. Multi user cursor.execute(f"ALTER DATABASE [{db_name}] SET MULTI_USER") output_lines.append("Base de datos configurada en modo MULTI_USER") elapsed = time.time() - start_time output_lines.append(f"Restauración completada en {elapsed:.2f}s") conn.close() output = "\n".join(output_lines) app_logger.info(output) return True, output, None except Exception as e: error_msg = f"Error restaurando base de datos: {str(e)}" app_logger.error(error_msg) return False, None, error_msg def database_exists(self, db_name: str) -> bool: """ Verifica si una base de datos existe. Args: db_name: Nombre de la base de datos Returns: True si existe """ try: conn_str = self.get_connection_string() conn = pyodbc.connect(conn_str, timeout=10) cursor = conn.cursor() cursor.execute( "SELECT COUNT(*) FROM sys.databases WHERE name = ?", (db_name,) ) exists = cursor.fetchone()[0] > 0 conn.close() return exists except Exception as e: app_logger.error(f"Error verificando existencia de DB: {e}") return False