"""Alembic helpers for PortfolioHub database migrations.""" from __future__ import annotations from pathlib import Path from alembic import command from alembic.config import Config import psycopg from app.core.config import Settings def get_alembic_config(settings: Settings) -> Config: project_root = Path(__file__).resolve().parents[2] config = Config(str(project_root / "alembic.ini")) config.set_main_option("script_location", str(project_root / "migrations")) config.set_main_option("sqlalchemy.url", settings.active_database_url.replace("%", "%%")) # migrations/env.py must use this explicit settings object instead of the # process-global cached Settings. This keeps programmatic callers (notably # isolated integration tests and runtime bootstrap) on their requested DB. config.attributes["portfoliohub_settings"] = settings return config def upgrade_to_head(settings: Settings) -> None: command.upgrade(get_alembic_config(settings), "head") def bootstrap_or_upgrade_to_head(settings: Settings) -> str: if _has_alembic_version_table(settings): upgrade_to_head(settings) return "head" if _has_legacy_portfolio_schema(settings): # Runtime ensure_schema() can create the BAT-025 columns before Alembic is # introduced. Stamp only the revision immediately before BAT-025, then run # the real upgrade so 0028 backfills a nullable pre-created `source` and # converges constraints/defaults without replaying the older DDL chain. config = get_alembic_config(settings) command.stamp(config, "20260628_0027") command.upgrade(config, "head") return "head" upgrade_to_head(settings) return "head" def current_revision(settings: Settings) -> str | None: output: list[str] = [] def capture(message: str, *_args, **_kwargs) -> None: if message.strip(): output.append(message.strip()) config = get_alembic_config(settings) command.current(config, verbose=False) return output[-1] if output else None def _connect(settings: Settings) -> psycopg.Connection: return psycopg.connect( host=settings.active_db_host, port=settings.active_db_port, dbname=settings.active_db_name, user=settings.active_db_user, password=settings.active_db_password, sslmode=settings.active_db_sslmode, connect_timeout=settings.active_db_connect_timeout_sec, application_name=settings.active_db_application_name, autocommit=True, ) def _has_alembic_version_table(settings: Settings) -> bool: with _connect(settings) as conn: with conn.cursor() as cur: cur.execute("SELECT to_regclass('public.alembic_version')") return cur.fetchone()[0] is not None def _has_legacy_portfolio_schema(settings: Settings) -> bool: required_tables = {"strategy_sets", "portfolios", "portfolio_ledger"} with _connect(settings) as conn: with conn.cursor() as cur: cur.execute( "SELECT tablename FROM pg_tables WHERE schemaname = 'public' AND tablename = ANY(%s)", (list(required_tables),), ) found = {row[0] for row in cur.fetchall()} return found == required_tables