Function run_migrations

hindsight-api-slim/hindsight_api/migrations.py:236–385

Run database migrations to the latest version using programmatic Alembic configuration. This function is safe to call from multiple distributed workers simultaneously: - Uses PostgreSQL advisory lock to ensure only one worker runs migrations at a time - Other workers wait for the l

(
    database_url: str,
    script_location: str | None = None,
    schema: str | None = None,
    migration_database_url: str | None = None,
)

def run_migrations(
    database_url: str,
    script_location: str | None = None,
    schema: str | None = None,
    migration_database_url: str | None = None,
) -> None:
    """
    Run database migrations to the latest version using programmatic Alembic configuration.

    This function is safe to call from multiple distributed workers simultaneously:
    - Uses PostgreSQL advisory lock to ensure only one worker runs migrations at a time
    - Other workers wait for the lock, then verify migrations are complete
    - If schema is already up-to-date, this is a fast no-op

    Supports multi-tenant schema isolation: when a schema is specified, migrations
    run in that schema instead of public. This allows tenant extensions to provision
    new tenant schemas with their own isolated tables.

    Args:
        database_url: SQLAlchemy database URL (e.g., "postgresql://user:pass@host/db")
        script_location: Path to alembic migrations directory (e.g., "/path/to/alembic").
                        If None, defaults to hindsight-api/alembic directory.
        schema: Target PostgreSQL schema name. If None, uses default (public).
                When specified, creates the schema if needed and runs migrations there.

    Raises:
        RuntimeError: If migrations fail to complete
        FileNotFoundError: If script_location doesn't exist

    Example:
        # Using default location and public schema
        run_migrations("postgresql://user:pass@host/db")

        # Run migrations for a specific tenant schema
        run_migrations("postgresql://user:pass@host/db", schema="tenant_acme")

        # Using custom location (when importing from another project)
        run_migrations(
            "postgresql://user:pass@host/db",
            script_location="/path/to/copied/_alembic"
        )
    """
    # Prefer a dedicated migration URL that bypasses connection poolers (e.g.
    # PgBouncer in transaction mode).  Session-level advisory locks don't
    # survive a PgBouncer transaction-mode cycle, so the distributed lock is
    # ineffective when the app URL goes through a pooler.  Configure
    # HINDSIGHT_API_MIGRATION_DATABASE_URL to the direct PostgreSQL endpoint
    # (e.g. hindsight-pg-rw) to restore correct locking behaviour.
    raw_url = migration_database_url or database_url
    # Oracle URLs are passed through to SQLAlchemy unchanged; only PG URLs
    # need the libpq normalization (asyncpg → psycopg2 driver, ssl → sslmode).
    migration_url = raw_url if is_oracle_url(raw_url) else to_libpq_url(raw_url)

    try:
        # Determine script location
        if script_location is None:
            # Default: use the alembic directory inside the hindsight_api package
            # This file is in: hindsight_api/migrations.py