plantillas-proyectos/backend/alembic/env.py

import importlib.util
import logging
import os
import sys
from logging.config import fileConfig
from urllib.parse import quote_plus

from alembic import context
from core.config import settings
from core.database import Base
from sqlalchemy import engine_from_config, pool

logger = logging.getLogger(__name__)

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config


def get_database_url():
    """Obtiene la URL de la base de datos (PostgreSQL) desde variables de entorno o alembic.ini."""
    # Intentar construir desde variables de entorno primero
    host = settings.CORE_DB_HOST
    db = settings.CORE_DB_NAME
    user = settings.CORE_DB_USER
    password = settings.CORE_DB_PASSWORD
    port = settings.CORE_DB_PORT

    if host and db and user and password:
        try:
            encoded_user = quote_plus(user)
            encoded_password = quote_plus(password)
            encoded_db = quote_plus(db)
            return f"postgresql+psycopg2://{encoded_user}:{encoded_password}@{host}:{port}/{encoded_db}"
        except Exception as e:
            logger.error(f"Error al construir URL: {e}")

    # Fallback al archivo de configuración
    url = config.get_main_option("sqlalchemy.url")
    if not url:
        raise RuntimeError(
            "No se ha configurado la cadena de conexión a PostgreSQL. "
            "Proporciona las variables de entorno POSTGRES_* o configura sqlalchemy.url en alembic.ini"
        )

    return url


# Configurar la URL de la base de datos
database_url = get_database_url()

# Debug: mostrar la URL (sin la contraseña)
if os.environ.get("ALEMBIC_DEBUG"):
    # Ocultar la contraseña para el debug en la URL
    try:
        before, after = database_url.split("@", 1)
        if ":" in before:
            before = before.split(":", 1)[0] + ":***"
        debug_url = before + "@" + after
    except Exception:
        debug_url = "postgresql://***:***@***"
        logger.error("Error al ocultar la contraseña en la URL para debug.")

config.set_main_option("sqlalchemy.url", database_url)

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Ajusta la ruta para que puedas importar core y módulos
BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, BASE_DIR)

# Configuración de Alembic
config = context.config
fileConfig(config.config_file_name)
target_metadata = Base.metadata


def import_models_from_dir(dir_path: str):
    """Importa recursivamente cualquier archivo models.py desde dir_path y archivos en directorios models/"""
    import sys

    def _load_module(module_path: str):
        rel_path = os.path.relpath(module_path, BASE_DIR)
        module_name = rel_path.replace(os.sep, ".").replace(".py", "")
        # Skip if already loaded to avoid duplicate SQLAlchemy table registrations
        if module_name in sys.modules:
            return
        spec = importlib.util.spec_from_file_location(module_name, module_path)
        mod = importlib.util.module_from_spec(spec)
        # Register in sys.modules before exec so transitive imports resolve correctly
        sys.modules[module_name] = mod
        spec.loader.exec_module(mod)

    for root, dirs, files in os.walk(dir_path):
        # Importar archivos models.py directos
        if "models.py" in files:
            try:
                _load_module(os.path.join(root, "models.py"))
            except Exception as e:
                logger.warning(f"No se pudo importar {os.path.join(root, 'models.py')}: {e}")

        # Importar todos los archivos .py en directorios llamados "models"
        if os.path.basename(root) == "models":
            for file in files:
                if file.endswith(".py") and not file.startswith("__"):
                    try:
                        _load_module(os.path.join(root, file))
                    except Exception as e:
                        logger.warning(f"No se pudo importar {os.path.join(root, file)}: {e}")


# Importar todos los models dentro de api/v1/modules y api/v1/modules/uploads
modules_dir = os.path.join(BASE_DIR, "api", "v1", "modules")
import_models_from_dir(modules_dir)


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.
    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.
    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()