file

FileMigrationRunner for executing file-based SQL 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()