auto

AutoMigrationRunner — generates and executes migrations from state diffs.

AutoMigrationRunner

Bases: MigrationRunner

Generate migrations from a state diff and execute them.

The runner compares a current state (read from the database statefile) against a desired state (parsed from SQL files) and generates the SQL migrations needed to reconcile the two. After execution the applied migrations are persisted to auto_migration_state_path for observability.

Source code in src/cloe_delta_table_manager/migrations/runner/auto.py

class AutoMigrationRunner(MigrationRunner):
    """Generate migrations from a state diff and execute them.

    The runner compares a *current* state (read from the database statefile)
    against a *desired* state (parsed from SQL files) and generates the SQL
    migrations needed to reconcile the two.  After execution the applied
    migrations are persisted to ``auto_migration_state_path`` for observability.
    """

    def __init__(
        self,
        *,
        executor: SQLExecutor | None = None,
        state_client: StateClient | None = None,
        platform_options: AzureDatabricksOptions | None = None,
    ) -> None:
        """Initialize the AutoMigrationRunner."""
        super().__init__(executor=executor, state_client=state_client)
        self._platform_options = platform_options

    def execute_migrations(  # type: ignore[override]
        self,
        *,
        current_state: DatabaseState,
        desired_state: DatabaseState,
        statefile_path: str | Path,
    ) -> None:
        """Generate and execute auto migrations, then persist state.

        Args:
            current_state: The database state as recorded in the statefile.
            desired_state: The desired state derived from SQL source files.
            statefile_path: Path to persist the auto migration state after execution.

        """
        self._generate_migrations(current_state, desired_state)
        if not self._state.migrations:
            logger.info("Auto migrations: nothing to do — states are in sync.")
            return
        logger.info(f"Auto migrations: {len(self._state.migrations)} migration(s) to apply.")
        self._run_migrations(statefile_path)

    def list_pending_migrations(
        self,
        *,
        current_state: DatabaseState,
        desired_state: DatabaseState,
    ) -> list[Migration]:
        """Return auto-generated migrations without executing them.

        Args:
            current_state: The database state as recorded in the statefile.
            desired_state: The desired state derived from SQL source files.

        Returns:
            List of generated Migration objects.

        """
        self._generate_migrations(current_state, desired_state)
        return self._state.migrations

    def _generate_migrations(
        self,
        current_state: DatabaseState,
        desired_state: DatabaseState,
    ) -> None:
        migrations = MigrationGenerator.generate_migrations(
            current_state=current_state,
            desired_state=desired_state,
            platform_options=self._platform_options,
        )
        self._state = MigrationState(migrations=migrations)

__init__(*, executor=None, state_client=None, platform_options=None)

Initialize the AutoMigrationRunner.

Source code in src/cloe_delta_table_manager/migrations/runner/auto.py

def __init__(
    self,
    *,
    executor: SQLExecutor | None = None,
    state_client: StateClient | None = None,
    platform_options: AzureDatabricksOptions | None = None,
) -> None:
    """Initialize the AutoMigrationRunner."""
    super().__init__(executor=executor, state_client=state_client)
    self._platform_options = platform_options

execute_migrations(*, current_state, desired_state, statefile_path)

Generate and execute auto migrations, then persist state.

Parameters:

Name	Type	Description	Default
current_state	DatabaseState	The database state as recorded in the statefile.	required
desired_state	DatabaseState	The desired state derived from SQL source files.	required
statefile_path	str | Path	Path to persist the auto migration state after execution.	required

Source code in src/cloe_delta_table_manager/migrations/runner/auto.py

def execute_migrations(  # type: ignore[override]
    self,
    *,
    current_state: DatabaseState,
    desired_state: DatabaseState,
    statefile_path: str | Path,
) -> None:
    """Generate and execute auto migrations, then persist state.

    Args:
        current_state: The database state as recorded in the statefile.
        desired_state: The desired state derived from SQL source files.
        statefile_path: Path to persist the auto migration state after execution.

    """
    self._generate_migrations(current_state, desired_state)
    if not self._state.migrations:
        logger.info("Auto migrations: nothing to do — states are in sync.")
        return
    logger.info(f"Auto migrations: {len(self._state.migrations)} migration(s) to apply.")
    self._run_migrations(statefile_path)

list_pending_migrations(*, current_state, desired_state)

Return auto-generated migrations without executing them.

Parameters:

Name	Type	Description	Default
current_state	DatabaseState	The database state as recorded in the statefile.	required
desired_state	DatabaseState	The desired state derived from SQL source files.	required

Returns:

Type	Description
list[Migration]	List of generated Migration objects.

Source code in src/cloe_delta_table_manager/migrations/runner/auto.py

def list_pending_migrations(
    self,
    *,
    current_state: DatabaseState,
    desired_state: DatabaseState,
) -> list[Migration]:
    """Return auto-generated migrations without executing them.

    Args:
        current_state: The database state as recorded in the statefile.
        desired_state: The desired state derived from SQL source files.

    Returns:
        List of generated Migration objects.

    """
    self._generate_migrations(current_state, desired_state)
    return self._state.migrations