hub-core/hub_core/migrations/roles.py

from __future__ import annotations

import re


def migration_role_statement(role: str | None) -> str | None:
    """Return a safely quoted SET ROLE statement for an admitted owner role."""
    if not role:
        return None
    if not re.fullmatch(r"[a-z_][a-z0-9_]{0,62}", role):
        raise ValueError("HUB_CORE_MIGRATION_ROLE is not a safe PostgreSQL role name")
    return f'SET ROLE "{role}"'


def migration_schema_statement(schema: str | None) -> str | None:
    """Return an explicit, safely quoted migration search path."""
    if not schema:
        return None
    if not re.fullmatch(r"[a-z_][a-z0-9_]{0,62}", schema):
        raise ValueError("HUB_CORE_MIGRATION_SCHEMA is not a safe PostgreSQL schema name")
    return f'SET search_path TO "{schema}", public'