278 lines
9.3 KiB
Python
278 lines
9.3 KiB
Python
"""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
|