migrate.py

import os
import subprocess
import sys
from dotenv import load_dotenv
from sqlalchemy import create_engine, inspect, text

# 1. Cargamos las variables de entorno
load_dotenv()

# Obtenemos las credenciales
DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_HOST = os.getenv("DB_HOST")
DB_PORT = os.getenv("DB_PORT")
DB_NAME = os.getenv("DB_NAME")

DATABASE_URL = f"postgresql+psycopg2://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"

def clean_orphan_alembic_version():
    """
    Detecta si la tabla 'alembic_version' existe en la base de datos pero no existen
    archivos de migración en la carpeta 'alembic/versions'. Si es así, la elimina
    para evitar conflictos de versiones huérfanas.
    """
    print("Verificando existencia de versiones huérfanas...")
    versions_dir = os.path.join("alembic", "versions")
    
    # Comprobar si hay archivos de migración (.py)
    has_migration_files = False
    if os.path.exists(versions_dir):
        files = [f for f in os.listdir(versions_dir) if f.endswith(".py")]
        if files:
            has_migration_files = True

    # Si no hay archivos locales pero la tabla alembic_version existe, la limpiamos
    if not has_migration_files:
        try:
            engine = create_engine(DATABASE_URL)
            inspector = inspect(engine)
            if "alembic_version" in inspector.get_table_names():
                print("Se detectó una tabla 'alembic_version' huérfana en la base de datos. Eliminándola...")
                with engine.begin() as conn:
                    conn.execute(text("DROP TABLE alembic_version;"))
                print("Tabla 'alembic_version' eliminada con éxito.")
        except Exception as e:
            print(f"Nota: No se pudo verificar/eliminar la tabla alembic_version: {e}")

def run_alembic_commands():
    """
    Ejecuta de manera automatizada la generación de la migración y su posterior aplicación.
    """
    print("Generando migración autogenerada...")
    # Buscamos el comando uvicorn/alembic en el entorno virtual
    # Usamos sys.executable para encontrar el binario de Python actual en el venv
    venv_dir = os.path.dirname(sys.executable)
    alembic_cmd = os.path.join(venv_dir, "alembic")
    if os.name == "nt":  # Windows
        alembic_cmd += ".exe"
    
    # 1. Crear revisión autogenerada
    revision_result = subprocess.run(
        [alembic_cmd, "revision", "--autogenerate", "-m", "Crear tabla libros"],
        capture_output=True,
        text=True
    )
    
    print(revision_result.stdout)
    if revision_result.returncode != 0:
        print("Error al generar la revisión de Alembic:")
        print(revision_result.stderr)
        return False
        
    print("Aplicando migración (upgrade head)...")
    # 2. Aplicar la migración a la base de datos
    upgrade_result = subprocess.run(
        [alembic_cmd, "upgrade", "head"],
        capture_output=True,
        text=True
    )
    
    print(upgrade_result.stdout)
    if upgrade_result.returncode != 0:
        print("Error al aplicar la migración de Alembic:")
        print(upgrade_result.stderr)
        return False
        
    print("¡Migraciones completadas exitosamente!")
    return True

if __name__ == "__main__":
    clean_orphan_alembic_version()
    run_alembic_commands()