Index

Runner subpackage for migration execution.

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

FileMigrationRunner

Bases: MigrationRunner

Manages reading, ordering, and executing user-defined database migrations.

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

class FileMigrationRunner(MigrationRunner):
    """Manages reading, ordering, and executing user-defined database migrations."""

    def __init__(self, executor: SQLExecutor | None = None, state_client: StateClient | None = None):
        """Initialize the FileMigrationRunner.

        Args:
            executor: SQL executor for running statements against Databricks.
            state_client: Client for reading/writing state. Defaults to LocalStateClient.

        """
        super().__init__(executor=executor, state_client=state_client)
        self._prev_state: MigrationState = MigrationState(migrations=[])

    def execute_migrations(
        self,
        statefile_path: Path | str,
        root_path: Path | str,
        glob_pattern: str,
        conflict_strategy: MigrationConflictStrategy | None = None,
    ) -> None:
        """Execute user-defined migrations."""
        self._prepare_migrations(statefile_path, root_path, glob_pattern, conflict_strategy)
        print(f"Executing migrations from path: {root_path}/{glob_pattern}")
        self._run_migrations(statefile_path)

    def list_pending_migrations(
        self,
        statefile_path: Path | str,
        root_path: Path | str,
        glob_pattern: str,
        conflict_strategy: MigrationConflictStrategy | None = None,
    ) -> list[Migration]:
        """Return pending migrations in execution order without running them.

        Args:
            statefile_path: Path to the migration state file.
            root_path: Path to directory containing migration SQL files.
            glob_pattern: Glob pattern to match migration files.
            conflict_strategy: Strategy for handling modified/removed migrations.

        Returns:
            List of NEW migrations in dependency-resolved execution order.

        """
        self._prepare_migrations(statefile_path, root_path, glob_pattern, conflict_strategy)
        return MigrationGraphService(self._state.migrations).get_execution_order()

    def _prepare_migrations(
        self,
        statefile_path: Path | str,
        root_path: Path | str,
        glob_pattern: str,
        conflict_strategy: MigrationConflictStrategy | None = None,
    ) -> None:
        self._read_current_migrations_from_path(root_path, glob_pattern=glob_pattern)
        self._read_prev_migrations_from_statefile(statefile_path)
        self._validate_prev_migration(conflict_strategy or RaiseOnConflict())
        self._update_current_migration_statuses()

    def _persist_state(self, statefile_path: Path | str) -> None:
        """Write the full state (file runner already holds the complete migration set)."""
        self._state.to_file(statefile_path, client=self._state_client)

    def _read_current_migrations_from_path(self, root_path: Path | str, glob_pattern: str) -> None:
        """Read all migration files from a directory path.

        Args:
            root_path: Path to directory containing migration SQL files.
            glob_pattern: Glob pattern to match migration files (e.g., '*.sql').

        Raises:
            ValueError: If errors occur while reading or resolving migrations.

        """
        if isinstance(root_path, str):
            root_path = Path(root_path)

        errors = []

        for migration_file in sorted(root_path.glob(glob_pattern)):
            try:
                migration = self._read_migration_from_file(migration_file, root_path=root_path)
                self._state.migrations.append(migration)
            except ValueError as e:
                errors.append((migration_file, e))
        if errors:
            error_messages = "\n###\n".join([f"{file}: {error}" for file, error in errors])
            raise ValueError(f"Errors occurred while reading migrations:\n{error_messages}")

        self._resolve_source_path_dependencies()

    def _read_prev_migrations_from_statefile(self, path: Path | str):
        """Read migrations from a log file.

        Args:
            path: Path to the migration log file.

        Raises:
            FileNotFoundError: If the state file does not exist. Run ``dtm init``
                to create empty state files before the first deployment.

        """
        self._prev_state = self._state_client.read_state(str(path), MigrationState)

    def _update_current_migration_statuses(self):
        for current_migration in self._state.migrations:
            for prev_migration in self._prev_state.migrations:
                if current_migration.operation_hash == prev_migration.operation_hash:
                    current_migration.status = prev_migration.status
                    break
            else:
                # If the migration was not found in the log, we consider it as new.
                current_migration.status = MigrationStatus.NEW

    def _validate_prev_migration(self, strategy: MigrationConflictStrategy) -> None:
        strategy.on_modified(self._find_modified_migrations())
        strategy.on_removed(self._find_removed_migrations())

    def _find_modified_migrations(self) -> list[Migration]:
        modified_migrations = []
        for current_migration in self._state.migrations:
            # Only UserDefinedMigrations have source_path
            if not isinstance(current_migration, FileMigration):
                continue

            for prev_migration in self._prev_state.migrations:
                if not isinstance(prev_migration, FileMigration):
                    continue

                if (
                    current_migration.source_path == prev_migration.source_path
                    and current_migration.operation_hash != prev_migration.operation_hash
                ):
                    modified_migrations.append(current_migration)
                    break
        return modified_migrations

    def _find_removed_migrations(self) -> list[Migration]:
        removed_migrations = []
        for prev_migration in self._prev_state.migrations:
            # Only UserDefinedMigrations have source_path
            if not isinstance(prev_migration, FileMigration):
                continue
            for current_migration in self._state.migrations:
                if not isinstance(current_migration, FileMigration):
                    continue

                if prev_migration.source_path == current_migration.source_path:
                    break
                removed_migrations.append(prev_migration)
        return removed_migrations

    def _resolve_source_path_dependencies(self):
        errors = []
        for migration in self.file_migrations:
            for dep_path in migration.depends_on_paths:
                dep_migration = next((m for m in self.file_migrations if m.source_path == dep_path), None)
                if dep_migration is None:
                    errors.append(
                        ValueError(
                            f"Migration {migration.source_path} depends on {dep_path}, which was not found among the loaded migrations."
                        )
                    )
                else:
                    migration.depends_on.append(dep_migration)

        if errors:
            error_messages = "\n###\n".join(str(error) for error in errors)
            raise ValueError(f"Errors occurred while resolving dependencies:\n{error_messages}")

    @staticmethod
    def _read_migration_from_file(migration_file: Path | str, root_path: Path | None = None):
        if isinstance(migration_file, str):
            migration_file = Path(migration_file)
        source_path: Path = migration_file.relative_to(root_path) if root_path else migration_file
        with migration_file.open("r") as f:
            sql = f.read()

        parsed_sql = sqlglot.parse_one(sql)

        try:
            comments = parsed_sql.comments
            if comments:
                first_comment = comments[0]
                migration_metadata = yaml.safe_load(first_comment)
            else:
                migration_metadata = {}
        except AttributeError, IndexError:
            migration_metadata = {}
        except yaml.YAMLError as e:
            raise ValueError(
                f"Error parsing YAML metadata in {migration_file}: {e}. "
                "The first comment in a migration file must be valid YAML metadata "
                "(e.g. 'depends_on'); generic leading comments are not supported."
            ) from e

        depends_on = migration_metadata.get("depends_on", [])
        depends_on_paths = depends_on if isinstance(depends_on, list) else [depends_on]

        return FileMigration(
            operations=[
                Operation(
                    commands=[sql],
                    operation_type=OperationType.USER,
                    index=0,
                )
            ],
            status=MigrationStatus.UNDEFINED,
            depends_on_paths=depends_on_paths,
            source_path=source_path,
        )

__init__(executor=None, state_client=None)

Initialize the FileMigrationRunner.

Parameters:

Name	Type	Description	Default
executor	SQLExecutor | None	SQL executor for running statements against Databricks.	None
state_client	StateClient | None	Client for reading/writing state. Defaults to LocalStateClient.	None

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

def __init__(self, executor: SQLExecutor | None = None, state_client: StateClient | None = None):
    """Initialize the FileMigrationRunner.

    Args:
        executor: SQL executor for running statements against Databricks.
        state_client: Client for reading/writing state. Defaults to LocalStateClient.

    """
    super().__init__(executor=executor, state_client=state_client)
    self._prev_state: MigrationState = MigrationState(migrations=[])

execute_migrations(statefile_path, root_path, glob_pattern, conflict_strategy=None)

Execute user-defined migrations.

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

def execute_migrations(
    self,
    statefile_path: Path | str,
    root_path: Path | str,
    glob_pattern: str,
    conflict_strategy: MigrationConflictStrategy | None = None,
) -> None:
    """Execute user-defined migrations."""
    self._prepare_migrations(statefile_path, root_path, glob_pattern, conflict_strategy)
    print(f"Executing migrations from path: {root_path}/{glob_pattern}")
    self._run_migrations(statefile_path)

list_pending_migrations(statefile_path, root_path, glob_pattern, conflict_strategy=None)

Return pending migrations in execution order without running them.

Parameters:

Name	Type	Description	Default
statefile_path	Path | str	Path to the migration state file.	required
root_path	Path | str	Path to directory containing migration SQL files.	required
glob_pattern	str	Glob pattern to match migration files.	required
conflict_strategy	MigrationConflictStrategy | None	Strategy for handling modified/removed migrations.	None

Returns:

Type	Description
list[Migration]	List of NEW migrations in dependency-resolved execution order.

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

def list_pending_migrations(
    self,
    statefile_path: Path | str,
    root_path: Path | str,
    glob_pattern: str,
    conflict_strategy: MigrationConflictStrategy | None = None,
) -> list[Migration]:
    """Return pending migrations in execution order without running them.

    Args:
        statefile_path: Path to the migration state file.
        root_path: Path to directory containing migration SQL files.
        glob_pattern: Glob pattern to match migration files.
        conflict_strategy: Strategy for handling modified/removed migrations.

    Returns:
        List of NEW migrations in dependency-resolved execution order.

    """
    self._prepare_migrations(statefile_path, root_path, glob_pattern, conflict_strategy)
    return MigrationGraphService(self._state.migrations).get_execution_order()

MigrationRunner

Bases: ABC

Manages reading, ordering, and executing database migrations.

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

class MigrationRunner(ABC):
    """Manages reading, ordering, and executing database migrations."""

    def __init__(self, executor: SQLExecutor | None = None, state_client: StateClient | None = None):
        """Initialize the MigrationRunner.

        Args:
            executor: SQL executor for running statements against Databricks.
            state_client: Client for reading/writing state. Defaults to LocalStateClient.

        """
        self._state: MigrationState = MigrationState(migrations=[])
        self._state_client: StateClient = state_client or LocalStateClient()
        self._executor: SQLExecutor | None = executor

    @property
    def file_migrations(self) -> list[FileMigration]:
        """Get list of file-based migrations only.

        Returns:
            List of FileMigration instances.

        """
        return [m for m in self._state.migrations if isinstance(m, FileMigration)]

    @property
    def pending_migrations(self) -> list[Migration]:
        """Get list of pending migrations.

        Returns:
            List of migrations that are in the PENDING state.

        """
        return [m for m in self._state.migrations if m.status == MigrationStatus.PENDING]

    @property
    def applied_migrations(self) -> list[Migration]:
        """Get list of applied migrations.

        Returns:
            List of migrations that are in the APPLIED state.

        """
        return [m for m in self._state.migrations if m.status == MigrationStatus.APPLIED]

    @property
    def new_migrations(self) -> list[Migration]:
        """Get list of new migrations.

        Returns:
            List of migrations that are in the NEW state.

        """
        return [m for m in self._state.migrations if m.status == MigrationStatus.NEW]

    @property
    def failed_migrations(self) -> list[Migration]:
        """Get list of failed migrations.

        Returns:
            List of migrations that are in the FAILED state.

        """
        return [m for m in self._state.migrations if m.status == MigrationStatus.FAILED]

    @abstractmethod
    def execute_migrations(self, *args, **kwargs) -> None:
        """Execute migrations."""

    def _execute_command(self, command: str, migration: Migration) -> bool:
        """Execute a single command. Return True if any statement failed."""
        assert self._executor is not None
        results = self._executor.execute_statements(command)
        failed_results = [r for r in results if r.failed]
        for r in failed_results:
            logger.error(f"Statement failed in migration {migration}: {r.error}")
        return bool(failed_results)

    def _execute_migration(self, migration: Migration) -> bool:
        """Execute all operations of a migration. Return True if it should be aborted."""
        try:
            for op in sorted(migration.operations, key=lambda op: op.index):
                logger.info(f"  {op}")
                for command in op.commands:
                    if self._execute_command(command, migration):
                        return True
        except Exception:
            logger.exception(f"Unexpected error while executing migration: {migration}")
            return True
        return False

    def _run_migrations(self, statefile_path: Path | str | None = None) -> None:
        """Run pending migrations against the database."""
        graph_service = MigrationGraphService(self._state.migrations)
        while (migration := graph_service.get_ready_migration()) is not None:
            if self._executor is None:
                raise RuntimeError("No SQL executor configured. Pass an executor when initialising the runner.")
            logger.info(f"Executing migration: {migration}")

            if self._execute_migration(migration):
                migration.status = MigrationStatus.FAILED
                logger.warning(f"Migration {migration} left as {migration.status.value}. Stopping further execution.")
                break

            migration.status = MigrationStatus.APPLIED
            logger.info(f"Migration applied: {migration}")

        if statefile_path:
            self._persist_state(statefile_path)

        if self.failed_migrations:
            raise RuntimeError(
                f"Migration phase aborted: {len(self.failed_migrations)} migration(s) failed. "
                "Subsequent phases will not run."
            )

    def _persist_state(self, statefile_path: Path | str) -> None:
        """Append executed migrations to the existing state file."""
        try:
            existing = MigrationState.from_file(statefile_path, client=self._state_client)
        except FileNotFoundError, ValueError:
            existing = MigrationState()

        existing.migrations.extend(self._state.migrations)
        existing.metadata = self._state.metadata
        existing.to_file(statefile_path, client=self._state_client)

applied_migrations property

Get list of applied migrations.

Returns:

Type	Description
list[Migration]	List of migrations that are in the APPLIED state.

failed_migrations property

Get list of failed migrations.

Returns:

Type	Description
list[Migration]	List of migrations that are in the FAILED state.

file_migrations property

Get list of file-based migrations only.

Returns:

Type	Description
list[FileMigration]	List of FileMigration instances.

new_migrations property

Get list of new migrations.

Returns:

Type	Description
list[Migration]	List of migrations that are in the NEW state.

pending_migrations property

Get list of pending migrations.

Returns:

Type	Description
list[Migration]	List of migrations that are in the PENDING state.

__init__(executor=None, state_client=None)

Initialize the MigrationRunner.

Parameters:

Name	Type	Description	Default
executor	SQLExecutor | None	SQL executor for running statements against Databricks.	None
state_client	StateClient | None	Client for reading/writing state. Defaults to LocalStateClient.	None

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

def __init__(self, executor: SQLExecutor | None = None, state_client: StateClient | None = None):
    """Initialize the MigrationRunner.

    Args:
        executor: SQL executor for running statements against Databricks.
        state_client: Client for reading/writing state. Defaults to LocalStateClient.

    """
    self._state: MigrationState = MigrationState(migrations=[])
    self._state_client: StateClient = state_client or LocalStateClient()
    self._executor: SQLExecutor | None = executor

execute_migrations(*args, **kwargs) abstractmethod

Execute migrations.

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

@abstractmethod
def execute_migrations(self, *args, **kwargs) -> None:
    """Execute migrations."""