Update app.py

app.py

@@ -1,5 +1,4 @@
 import os
-import uuid
 import io
 import zipfile
 import re
@@ -17,8 +16,7 @@ from langdetect import detect
 from transformers import MarianMTModel, MarianTokenizer
 from openai import OpenAI
 
-import psycopg2
-import mysql.connector
+from sqlmanager import SQLManager
 
 # ======================================================
 # 0) Configuración general
@@ -28,25 +26,8 @@ import mysql.connector
 MODEL_DIR = os.getenv("MODEL_DIR", "stvnnnnnn/t5-large-nl2sql-spider")
 DEVICE = torch.device("cpu")  # inferencia en CPU
 
-# === Motores reales: variables de entorno ===
-# PostgreSQL: usaremos UN solo DB (POSTGRES_DB) y un schema por conexión lógica
-POSTGRES_HOST = os.getenv("POSTGRES_HOST", "localhost")
-POSTGRES_PORT = int(os.getenv("POSTGRES_PORT", "5432"))
-POSTGRES_USER = os.getenv("POSTGRES_USER", "postgres")
-POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "postgres")
-POSTGRES_DB = os.getenv("POSTGRES_DB", "postgres")
-
-# MySQL: crearemos una base de datos por conexión lógica
-MYSQL_HOST = os.getenv("MYSQL_HOST", "localhost")
-MYSQL_PORT = int(os.getenv("MYSQL_PORT", "3306"))
-MYSQL_USER = os.getenv("MYSQL_USER", "root")
-MYSQL_PASSWORD = os.getenv("MYSQL_PASSWORD", "root")
-
-# Registro en memoria de conexiones:
-# { conn_id: { "engine": "postgresql"|"mysql", "namespace": str, "label": str } }
-#   - engine = motor real
-#   - namespace = schema (Postgres) o database (MySQL) donde vive esa BD
-DB_REGISTRY: Dict[str, Dict[str, Any]] = {}
+# Gestor de conexiones reales (MySQL/PostgreSQL)
+sql_manager = SQLManager()
 
 # Cliente OpenAI para transcripción de audio (Whisper / gpt-4o-transcribe)
 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
@@ -62,8 +43,8 @@ app = FastAPI(
     title="NL2SQL T5-large Backend (MySQL/PostgreSQL)",
     description=(
         "Intérprete NL→SQL (T5-large Spider) para usuarios no expertos. "
-        "El usuario sube dumps .sql / .zip y se cargan en motores reales "
-        "(MySQL/PostgreSQL)."
+        "El usuario sube sus dumps .sql (o ZIP con .sql) y se levantan "
+        "bases reales en MySQL/PostgreSQL; las consultas se ejecutan ahí."
     ),
     version="2.0.0",
 )
@@ -137,271 +118,12 @@ def translate_es_to_en(text: str) -> str:
 
 
 # ======================================================
-# 3) Conexiones a motores reales y helpers
-# ======================================================
-
-def get_pg_conn():
-    return psycopg2.connect(
-        host=POSTGRES_HOST,
-        port=POSTGRES_PORT,
-        user=POSTGRES_USER,
-        password=POSTGRES_PASSWORD,
-        dbname=POSTGRES_DB,
-    )
-
-
-def get_mysql_conn(db_name: Optional[str] = None):
-    params = dict(
-        host=MYSQL_HOST,
-        port=MYSQL_PORT,
-        user=MYSQL_USER,
-        password=MYSQL_PASSWORD,
-    )
-    if db_name:
-        params["database"] = db_name
-    return mysql.connector.connect(**params)
-
-
-def detect_sql_dialect(sql_text: str) -> str:
-    """
-    Heurística simple:
-    - Si ve ENGINE, AUTO_INCREMENT, backticks, etc. → MySQL
-    - Si ve SERIAL, search_path, ::, PL/pgSQL, etc. → PostgreSQL
-    """
-    text = sql_text.lower()
-    if any(kw in text for kw in ["engine=", "auto_increment", "unsigned", " collate ", " character set ", "`"]):
-        return "mysql"
-    if any(kw in text for kw in ["serial", " set search_path", "copy ", "::", "language plpgsql"]):
-        return "postgresql"
-    return "unknown"
-
-
-def create_logical_db(engine: str, label: str) -> str:
-    """
-    Crea una "BD lógica":
-      - En PostgreSQL: un SCHEMA nuevo dentro de POSTGRES_DB
-      - En MySQL: una DATABASE nueva
-    """
-    conn_id = f"db_{uuid.uuid4().hex[:8]}"
-
-    if engine == "postgresql":
-        schema_name = conn_id
-        conn = get_pg_conn()
-        try:
-            conn.autocommit = True
-            cur = conn.cursor()
-            cur.execute(f'CREATE SCHEMA IF NOT EXISTS "{schema_name}";')
-            cur.close()
-        finally:
-            conn.close()
-        DB_REGISTRY[conn_id] = {"engine": "postgresql", "namespace": schema_name, "label": label}
-
-    elif engine == "mysql":
-        db_name = conn_id
-        conn = get_mysql_conn()
-        try:
-            cur = conn.cursor()
-            cur.execute(f"CREATE DATABASE IF NOT EXISTS `{db_name}`;")
-            conn.commit()
-            cur.close()
-        finally:
-            conn.close()
-        DB_REGISTRY[conn_id] = {"engine": "mysql", "namespace": db_name, "label": label}
-
-    else:
-        raise ValueError(f"Engine no soportado: {engine}")
-
-    return conn_id
-
-
-def ensure_connection(conn_id: str) -> Dict[str, Any]:
-    if conn_id not in DB_REGISTRY:
-        raise HTTPException(status_code=404, detail=f"connection_id '{conn_id}' no registrado")
-    return DB_REGISTRY[conn_id]
-
-
-# ======================================================
-# 4) Carga de scripts SQL (schema.sql / data.sql / ZIP)
-# ======================================================
-
-def _execute_sql_script_postgres(schema: str, sql_text: str) -> None:
-    conn = get_pg_conn()
-    try:
-        conn.autocommit = False
-        cur = conn.cursor()
-        # Trabajamos siempre dentro del schema lógico
-        cur.execute(f'SET search_path TO "{schema}";')
-        # Ejecución simple por ';' (suficiente para Sakila/Pagila)
-        parts = sql_text.split(";")
-        for stmt in parts:
-            s = stmt.strip()
-            if not s:
-                continue
-            cur.execute(s + ";")
-        conn.commit()
-        cur.close()
-    finally:
-        conn.close()
-
-
-def _execute_sql_script_mysql(db_name: str, sql_text: str) -> None:
-    conn = get_mysql_conn()
-    try:
-        cur = conn.cursor()
-        cur.execute(f"USE `{db_name}`;")
-        # mysql-connector permite multi=True
-        for _ in cur.execute(sql_text, multi=True):
-            pass
-        conn.commit()
-        cur.close()
-    finally:
-        conn.close()
-
-
-def load_sql_into_connection(sql_text: str, engine: str, conn_id: Optional[str], label: str) -> str:
-    """
-    - Si conn_id es None → crea nueva BD lógica y carga el script.
-    - Si conn_id existe → ejecuta el script encima (ej: primero schema.sql, luego data.sql).
-    """
-    if not engine or engine not in ("mysql", "postgresql"):
-        raise HTTPException(status_code=400, detail="Engine inválido o no soportado (usa 'mysql' o 'postgresql').")
-
-    if conn_id is None:
-        conn_id = create_logical_db(engine, label)
-    else:
-        ensure_connection(conn_id)
-
-    info = DB_REGISTRY[conn_id]
-    namespace = info["namespace"]
-
-    if engine == "postgresql":
-        _execute_sql_script_postgres(namespace, sql_text)
-    else:
-        _execute_sql_script_mysql(namespace, sql_text)
-
-    return conn_id
-
-
-# ======================================================
-# 5) Introspección de esquema y ejecución (sobre motores reales)
-# ======================================================
-
-def introspect_schema(conn_id: str) -> Dict[str, Any]:
-    info = ensure_connection(conn_id)
-    engine = info["engine"]
-    ns = info["namespace"]
-
-    tables_info: Dict[str, Dict[str, List[str]]] = {}
-    parts: List[str] = []
-
-    if engine == "postgresql":
-        conn = get_pg_conn()
-        try:
-            cur = conn.cursor()
-            cur.execute(
-                """
-                SELECT table_name, column_name
-                FROM information_schema.columns
-                WHERE table_schema = %s
-                ORDER BY table_name, ordinal_position;
-                """,
-                (ns,),
-            )
-            rows = cur.fetchall()
-            cur.close()
-        finally:
-            conn.close()
-        for table, col in rows:
-            tables_info.setdefault(table, {"columns": []})
-            tables_info[table]["columns"].append(col)
-
-    else:  # MySQL
-        conn = get_mysql_conn(ns)
-        try:
-            cur = conn.cursor()
-            cur.execute(
-                """
-                SELECT table_name, column_name
-                FROM information_schema.columns
-                WHERE table_schema = %s
-                ORDER BY table_name, ordinal_position;
-                """,
-                (ns,),
-            )
-            rows = cur.fetchall()
-            cur.close()
-        finally:
-            conn.close()
-        for table, col in rows:
-            tables_info.setdefault(table, {"columns": []})
-            tables_info[table]["columns"].append(col)
-
-    for t, info_t in tables_info.items():
-        parts.append(f"{t}(" + ", ".join(info_t["columns"]) + ")")
-
-    schema_str = " ; ".join(parts) if parts else "(empty_schema)"
-
-    return {
-        "tables": tables_info,
-        "foreign_keys": [],    # podrías enriquecer esto luego
-        "schema_str": schema_str,
-    }
-
-
-def execute_sql(conn_id: str, sql: str) -> Dict[str, Any]:
-    """
-    Ejecuta SOLO consultas de lectura (SELECT).
-    Bloquea operaciones destructivas por seguridad en el demo.
-    """
-    info = ensure_connection(conn_id)
-    engine = info["engine"]
-    ns = info["namespace"]
-
-    forbidden = ["drop ", "delete ", "update ", "insert ", "alter ", "replace ", "truncate ", "create "]
-    sql_low = sql.lower()
-    if any(f in sql_low for f in forbidden):
-        return {
-            "ok": False,
-            "error": "Query bloqueada por seguridad (operación potencialmente destructiva).",
-            "rows": None,
-            "columns": [],
-        }
-
-    try:
-        if engine == "postgresql":
-            conn = get_pg_conn()
-            try:
-                cur = conn.cursor()
-                cur.execute(f'SET search_path TO "{ns}";')
-                cur.execute(sql)
-                rows = cur.fetchall()
-                cols = [desc[0] for desc in cur.description] if cur.description else []
-                cur.close()
-            finally:
-                conn.close()
-        else:
-            conn = get_mysql_conn(ns)
-            try:
-                cur = conn.cursor()
-                cur.execute(sql)
-                rows = cur.fetchall()
-                cols = [desc[0] for desc in cur.description] if cur.description else []
-                cur.close()
-            finally:
-                conn.close()
-
-        return {"ok": True, "error": None, "rows": rows, "columns": cols}
-    except Exception as e:
-        return {"ok": False, "error": str(e), "rows": None, "columns": []}
-
-
-# ======================================================
-# 6) SQL REPAIR LAYER (igual que antes, pero agnóstico de motor)
+# 3) Capa de reparación de SQL (usa el schema real)
 # ======================================================
 
 def _normalize_name_for_match(name: str) -> str:
     s = name.lower()
-    s = s.replace('"', "").replace("`", "")
+    s = s.replace('"', '').replace("`", "")
     s = s.replace("_", "")
     if s.endswith("s") and len(s) > 3:
         s = s[:-1]
@@ -430,9 +152,11 @@ def _build_schema_indexes(tables_info: Dict[str, Dict[str, List[str]]]) -> Dict[
 def _best_match_name(missing: str, index: Dict[str, List[str]]) -> Optional[str]:
     if not index:
         return None
+
     key = _normalize_name_for_match(missing)
     if key in index and index[key]:
         return index[key][0]
+
     candidates = difflib.get_close_matches(key, list(index.keys()), n=1, cutoff=0.7)
     if not candidates:
         return None
@@ -465,6 +189,9 @@ DOMAIN_SYNONYMS_COLUMN = {
 
 
 def try_repair_sql(sql: str, error: str, schema_meta: Dict[str, Any]) -> Optional[str]:
+    """
+    Intenta reparar nombres de tablas/columnas basándose en el esquema real.
+    """
     tables_info = schema_meta["tables"]
     idx = _build_schema_indexes(tables_info)
     table_index = idx["table_index"]
@@ -476,11 +203,15 @@ def try_repair_sql(sql: str, error: str, schema_meta: Dict[str, Any]) -> Optiona
     missing_table = None
     missing_column = None
 
-    m_t = re.search(r"no such table: ([\w\.]+)", error or "", re.IGNORECASE)
+    m_t = re.search(r"relation \"([\w\.]+)\" does not exist", error, re.IGNORECASE)
+    if not m_t:
+        m_t = re.search(r"no such table: ([\w\.]+)", error)
     if m_t:
         missing_table = m_t.group(1)
 
-    m_c = re.search(r"no such column: ([\w\.]+)", error or "", re.IGNORECASE)
+    m_c = re.search(r"column \"([\w\.]+)\" does not exist", error, re.IGNORECASE)
+    if not m_c:
+        m_c = re.search(r"no such column: ([\w\.]+)", error)
     if m_c:
         missing_column = m_c.group(1)
 
@@ -522,7 +253,7 @@ def try_repair_sql(sql: str, error: str, schema_meta: Dict[str, Any]) -> Optiona
 
 
 # ======================================================
-# 7) Construcción de prompt y NL→SQL + re-ranking
+# 4) Construcción de prompt y NL→SQL + re-ranking
 # ======================================================
 
 def build_prompt(question_en: str, db_id: str, schema_str: str) -> str:
@@ -534,9 +265,18 @@ def build_prompt(question_en: str, db_id: str, schema_str: str) -> str:
 
 
 def nl2sql_with_rerank(question: str, conn_id: str) -> Dict[str, Any]:
-    ensure_connection(conn_id)
-    meta = introspect_schema(conn_id)
-    schema_str = meta["schema_str"]
+    if conn_id not in sql_manager.connections:
+        raise HTTPException(status_code=404, detail=f"connection_id '{conn_id}' no registrado")
+
+    # Obtener esquema real desde MySQL/Postgres
+    meta = sql_manager.get_schema(conn_id)
+    tables_info = meta["tables"]
+
+    parts = []
+    for t, info in tables_info.items():
+        cols = info.get("columns", [])
+        parts.append(f"{t}(" + ", ".join(cols) + ")")
+    schema_str = " ; ".join(parts) if parts else "(empty_schema)"
 
     detected = detect_language(question)
     question_en = translate_es_to_en(question) if detected == "es" else question
@@ -582,19 +322,21 @@ def nl2sql_with_rerank(question: str, conn_id: str) -> Dict[str, Any]:
             "raw_sql_model": raw_sql,
         }
 
-        exec_info = execute_sql(conn_id, raw_sql)
+        exec_info = sql_manager.execute_sql(conn_id, raw_sql)
 
+        # Intentar reparación solo si es error por tabla/columna
         if (not exec_info["ok"]) and (
             "no such table" in (exec_info["error"] or "").lower()
             or "no such column" in (exec_info["error"] or "").lower()
+            or "does not exist" in (exec_info["error"] or "").lower()
         ):
             current_sql = raw_sql
-            last_error = exec_info["error"]
+            last_error = exec_info["error"] or ""
             for step in range(1, 4):
-                repaired_sql = try_repair_sql(current_sql, last_error or "", meta)
+                repaired_sql = try_repair_sql(current_sql, last_error, meta)
                 if not repaired_sql or repaired_sql == current_sql:
                     break
-                exec_info2 = execute_sql(conn_id, repaired_sql)
+                exec_info2 = sql_manager.execute_sql(conn_id, repaired_sql)
                 cand["repaired_from"] = current_sql if cand["repaired_from"] is None else cand["repaired_from"]
                 cand["repair_note"] = f"auto-repair (table/column name, step {step})"
                 cand["sql"] = repaired_sql
@@ -602,12 +344,12 @@ def nl2sql_with_rerank(question: str, conn_id: str) -> Dict[str, Any]:
                 current_sql = repaired_sql
                 if exec_info2["ok"]:
                     break
-                last_error = exec_info2["error"]
+                last_error = exec_info2["error"] or ""
 
         cand["exec_ok"] = exec_info["ok"]
         cand["exec_error"] = exec_info["error"]
         cand["rows_preview"] = (
-            [list(r) for r in exec_info["rows"][:5]] if exec_info["ok"] and exec_info["rows"] else None
+            exec_info["rows"][:5] if exec_info["ok"] and exec_info["rows"] else None
         )
         cand["columns"] = exec_info["columns"]
 
@@ -641,33 +383,31 @@ def nl2sql_with_rerank(question: str, conn_id: str) -> Dict[str, Any]:
 
 
 # ======================================================
-# 8) Schemas Pydantic
+# 5) Schemas Pydantic
 # ======================================================
 
 class UploadResponse(BaseModel):
     connection_id: str
     label: str
-    engine: str
-    namespace: str  # schema (PG) o database (MySQL)
+    db_path: str  # ahora será un pseudo-path (engine://db_name)
     note: Optional[str] = None
 
 
 class ConnectionInfo(BaseModel):
     connection_id: str
     label: str
-    engine: str
+    engine: Optional[str] = None
+    db_name: Optional[str] = None
 
 
 class SchemaResponse(BaseModel):
     connection_id: str
-    engine: str
     schema_summary: str
     tables: Dict[str, Dict[str, List[str]]]
 
 
 class PreviewResponse(BaseModel):
     connection_id: str
-    engine: str
     table: str
     columns: List[str]
     rows: List[List[Any]]
@@ -698,27 +438,61 @@ class SpeechInferResponse(BaseModel):
 
 
 # ======================================================
-# 9) Endpoints FastAPI
+# 6) Helpers para /upload (.sql y .zip)
+# ======================================================
+
+def _combine_sql_files_from_zip(zip_bytes: bytes) -> str:
+    """
+    Lee un ZIP, se queda solo con los .sql y los concatena.
+    Orden:
+      1) archivos con 'schema' o 'structure' en el nombre
+      2) el resto (data, etc.)
+    """
+    try:
+        with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
+            names = [info.filename for info in zf.infolist() if not info.is_dir()]
+            sql_names = [n for n in names if n.lower().endswith(".sql")]
+
+            if not sql_names:
+                raise ValueError("El ZIP no contiene archivos .sql utilizables.")
+
+            def sort_key(name: str) -> int:
+                nl = name.lower()
+                if "schema" in nl or "structure" in nl:
+                    return 0
+                return 1
+
+            sql_names_sorted = sorted(sql_names, key=sort_key)
+
+            parts: List[str] = []
+            for name in sql_names_sorted:
+                with zf.open(name) as f:
+                    text = f.read().decode("utf-8", errors="ignore")
+                parts.append(f"-- FILE: {name}\n{text}\n")
+
+            return "\n\n".join(parts)
+    except zipfile.BadZipFile:
+        raise ValueError("Archivo ZIP inválido o corrupto.")
+
+
+# ======================================================
+# 7) Endpoints FastAPI
 # ======================================================
 
 @app.on_event("startup")
 async def startup_event():
     load_nl2sql_model()
-    print("✅ Backend NL2SQL inicializado (motores MySQL/PostgreSQL).")
+    print("✅ Backend NL2SQL inicializado (MySQL/PostgreSQL).")
+    print(f"MODEL_DIR={MODEL_DIR}, DEVICE={DEVICE}")
+    print(f"Conexiones activas al inicio: {len(sql_manager.connections)}")
 
 
-@app.post("/upload-sql", response_model=UploadResponse)
-async def upload_sql(
-    db_file: UploadFile = File(...),
-    engine: Optional[str] = Form(None),        # "mysql" | "postgresql" (opcional)
-    connection_id: Optional[str] = Form(None), # para el caso schema.sql + data.sql
-):
+@app.post("/upload", response_model=UploadResponse)
+async def upload_database(db_file: UploadFile = File(...)):
     """
-    Subida de dumps SQL.
-    - Acepta .sql directo o .zip con varios .sql
-    - Detecta motor automáticamente (MySQL/PostgreSQL) si engine no se especifica
-    - Si connection_id es None, crea una nueva BD lógica
-    - Si connection_id existe, ejecuta el SQL encima (ej: schema.sql luego data.sql)
+    Subida de BD basada en dumps:
+      - .sql → dump MySQL/PostgreSQL (schema + data) → BD real
+      - .zip → debe contener uno o varios .sql (se concatenan)
     """
     filename = db_file.filename
     if not filename:
@@ -727,140 +501,118 @@ async def upload_sql(
     fname_lower = filename.lower()
     contents = await db_file.read()
 
-    if not (fname_lower.endswith(".sql") or fname_lower.endswith(".zip")):
-        raise HTTPException(
-            status_code=400,
-            detail="Formato no soportado. Usa: .sql o .zip (con archivos .sql).",
-        )
-
-    sql_text = ""
-    note = None
+    note: Optional[str] = None
 
-    # Caso: archivo .sql único
+    # Caso 1: dump .sql
     if fname_lower.endswith(".sql"):
         sql_text = contents.decode("utf-8", errors="ignore")
-        detected = detect_sql_dialect(sql_text)
-        final_engine = engine or detected
-        if final_engine == "unknown":
+        try:
+            conn_id = sql_manager.create_database_from_dump(label=filename, sql_text=sql_text)
+        except Exception as e:
             raise HTTPException(
                 status_code=400,
-                detail=(
-                    "No se pudo detectar el motor SQL (MySQL/PostgreSQL). "
-                    "Vuelve a subir el archivo indicando engine='mysql' o 'postgresql'."
-                ),
+                detail=f"No se pudo crear la BD desde el dump SQL: {e}",
             )
+        meta = sql_manager.connections[conn_id]
+        engine = meta["engine"]
+        db_name = meta["db_name"]
+        note = f"SQL dump imported into {engine.upper()} database '{db_name}'."
 
-        conn_id = load_sql_into_connection(sql_text, final_engine, connection_id, filename)
-        info = DB_REGISTRY[conn_id]
-        note = f"SQL ejecutado sobre motor {final_engine}."
-        return UploadResponse(
-            connection_id=conn_id,
-            label=info["label"],
-            engine=info["engine"],
-            namespace=info["namespace"],
-            note=note,
-        )
-
-    # Caso: ZIP con varios .sql (ej: schema.sql + data.sql, o muchos scripts)
-    try:
-        with zipfile.ZipFile(io.BytesIO(contents)) as zf:
-            sql_names = [n for n in zf.namelist() if n.lower().endswith(".sql")]
-            if not sql_names:
-                raise HTTPException(
-                    status_code=400,
-                    detail="El ZIP no contiene archivos .sql utilizables.",
-                )
-
-            combined_sql_parts = []
-            for name in sorted(sql_names):
-                with zf.open(name) as f:
-                    combined_sql_parts.append(f"-- FILE: {name}\n")
-                    combined_sql_parts.append(f.read().decode("utf-8", errors="ignore"))
-            sql_text = "\n\n".join(combined_sql_parts)
+    # Caso 2: ZIP con uno o varios .sql
+    elif fname_lower.endswith(".zip"):
+        try:
+            sql_text = _combine_sql_files_from_zip(contents)
+        except ValueError as ve:
+            raise HTTPException(status_code=400, detail=str(ve))
 
-    except zipfile.BadZipFile:
-        raise HTTPException(status_code=400, detail="Archivo ZIP inválido o corrupto.")
+        try:
+            conn_id = sql_manager.create_database_from_dump(label=filename, sql_text=sql_text)
+        except Exception as e:
+            raise HTTPException(
+                status_code=400,
+                detail=f"No se pudo crear la BD desde los .sql dentro del ZIP: {e}",
+            )
+        meta = sql_manager.connections[conn_id]
+        engine = meta["engine"]
+        db_name = meta["db_name"]
+        note = f"ZIP with SQL dumps imported into {engine.upper()} database '{db_name}'."
 
-    detected = detect_sql_dialect(sql_text)
-    final_engine = engine or detected
-    if final_engine == "unknown":
+    else:
         raise HTTPException(
             status_code=400,
-            detail=(
-                "No se pudo detectar el motor SQL (MySQL/PostgreSQL) en el ZIP. "
-                "Vuelve a subir indicando engine='mysql' o 'postgresql'."
-            ),
+            detail="Formato no soportado. Usa: .sql o .zip (con archivos .sql dentro).",
         )
 
-    conn_id = load_sql_into_connection(sql_text, final_engine, connection_id, filename)
-    info = DB_REGISTRY[conn_id]
-    note = f"ZIP con scripts SQL ejecutado sobre motor {final_engine}."
+    meta = sql_manager.connections[conn_id]
+    engine = meta["engine"]
+    db_name = meta["db_name"]
+
+    # db_path ahora es un pseudo-path para mantener compatibilidad
+    db_path = f"{engine}://{db_name}"
+
     return UploadResponse(
         connection_id=conn_id,
-        label=info["label"],
-        engine=info["engine"],
-        namespace=info["namespace"],
+        label=meta["label"],
+        db_path=db_path,
         note=note,
     )
 
 
 @app.get("/connections", response_model=List[ConnectionInfo])
 async def list_connections():
-    out: List[ConnectionInfo] = []
-    for cid, info in DB_REGISTRY.items():
-        out.append(ConnectionInfo(connection_id=cid, label=info["label"], engine=info["engine"]))
-    return out
+    return [
+        ConnectionInfo(
+            connection_id=c["connection_id"],
+            label=c.get("label", ""),
+            engine=c.get("engine"),
+            db_name=c.get("db_name"),
+        )
+        for c in [
+            {
+                "connection_id": cid,
+                **meta,
+            }
+            for cid, meta in sql_manager.connections.items()
+        ]
+    ]
 
 
 @app.get("/schema/{connection_id}", response_model=SchemaResponse)
 async def get_schema(connection_id: str):
-    info = ensure_connection(connection_id)
-    meta = introspect_schema(connection_id)
+    if connection_id not in sql_manager.connections:
+        raise HTTPException(status_code=404, detail="connection_id no encontrado")
+
+    meta = sql_manager.get_schema(connection_id)
+    tables = meta["tables"]
+
+    parts = []
+    for t, info in tables.items():
+        cols = info.get("columns", [])
+        parts.append(f"{t}(" + ", ".join(cols) + ")")
+    schema_str = " ; ".join(parts) if parts else "(empty_schema)"
+
     return SchemaResponse(
         connection_id=connection_id,
-        engine=info["engine"],
-        schema_summary=meta["schema_str"],
-        tables=meta["tables"],
+        schema_summary=schema_str,
+        tables=tables,
     )
 
 
 @app.get("/preview/{connection_id}/{table}", response_model=PreviewResponse)
 async def preview_table(connection_id: str, table: str, limit: int = 20):
-    info = ensure_connection(connection_id)
-    engine = info["engine"]
-    ns = info["namespace"]
+    if connection_id not in sql_manager.connections:
+        raise HTTPException(status_code=404, detail="connection_id no encontrado")
 
     try:
-        if engine == "postgresql":
-            conn = get_pg_conn()
-            try:
-                cur = conn.cursor()
-                cur.execute(f'SET search_path TO "{ns}";')
-                cur.execute(f'SELECT * FROM "{table}" LIMIT %s;', (int(limit),))
-                rows = cur.fetchall()
-                cols = [d[0] for d in cur.description] if cur.description else []
-                cur.close()
-            finally:
-                conn.close()
-        else:
-            conn = get_mysql_conn(ns)
-            try:
-                cur = conn.cursor()
-                cur.execute(f"SELECT * FROM `{table}` LIMIT %s;", (int(limit),))
-                rows = cur.fetchall()
-                cols = [d[0] for d in cur.description] if cur.description else []
-                cur.close()
-            finally:
-                conn.close()
+        preview = sql_manager.get_preview(connection_id, table, limit)
     except Exception as e:
         raise HTTPException(status_code=400, detail=f"Error al leer tabla '{table}': {e}")
 
     return PreviewResponse(
         connection_id=connection_id,
-        engine=engine,
         table=table,
-        columns=cols,
-        rows=[list(r) for r in rows],
+        columns=preview["columns"],
+        rows=preview["rows"],
     )
 
 
@@ -873,12 +625,12 @@ async def infer_sql(req: InferRequest):
 @app.post("/speech-infer", response_model=SpeechInferResponse)
 async def speech_infer(
     connection_id: str = Form(...),
-    audio: UploadFile = File(...),
+    audio: UploadFile = File(...)
 ):
     if openai_client is None:
         raise HTTPException(
             status_code=500,
-            detail="OPENAI_API_KEY no está configurado en el backend.",
+            detail="OPENAI_API_KEY no está configurado en el backend."
         )
 
     if audio.content_type is None:
@@ -915,8 +667,7 @@ async def health():
     return {
         "status": "ok",
         "model_loaded": t5_model is not None,
-        "connections": len(DB_REGISTRY),
-        "engines_in_use": list({info["engine"] for info in DB_REGISTRY.values()}),
+        "connections": len(sql_manager.connections),
         "device": str(DEVICE),
     }
 
@@ -924,13 +675,13 @@ async def health():
 @app.get("/")
 async def root():
     return {
-        "message": "NL2SQL T5-large backend is running (MySQL/PostgreSQL engines).",
+        "message": "NL2SQL T5-large backend is running with real MySQL/PostgreSQL engines.",
         "endpoints": [
-            "POST /upload-sql        (subir .sql o .zip con .sql, engine auto o manual)",
-            "GET  /connections       (listar BDs lógicas)",
-            "GET  /schema/{id}       (esquema resumido desde motor real)",
+            "POST /upload            (subir .sql o .zip con .sql → crear BD dinámica)",
+            "GET  /connections       (listar BDs subidas)",
+            "GET  /schema/{id}       (esquema resumido)",
             "GET  /preview/{id}/{t}  (preview de tabla)",
-            "POST /infer             (NL→SQL + ejecución segura en el motor)",
+            "POST /infer             (NL→SQL + ejecución en BD real)",
             "POST /speech-infer      (NL por voz → SQL + ejecución)",
             "GET  /health            (estado del backend)",
             "GET  /docs              (OpenAPI UI)",