migration_orchestrator

Orchestrate the execution of pre-deployment, auto, and post-deployment migrations.

MigrationOrchestrator

Orchestrates the migration process by coordinating loading and running of migrations.

Source code in src/cloe_delta_table_manager/migrations/migration_orchestrator.py

class MigrationOrchestrator:
    """Orchestrates the migration process by coordinating loading and running of migrations."""

    def __init__(
        self,
        config: MigrationOrchestratorConfig,
        target_platform: Literal["azure_databricks"] = "azure_databricks",
        connector: DatabricksConnector | None = None,
        state_client: StateClient | None = None,
        platform_options: AzureDatabricksOptions | None = None,
    ):
        """Initialize the MigrationOrchestrator.

        Args:
            config: Migration paths and state file configuration.
            target_platform: Target platform for migrations (for now, only "azure_databricks" is supported).
            state_client: Client for reading/writing state. Defaults to LocalStateClient.
            connector: Databricks connector. Required to call ``deploy()``; may be omitted
                when only calling ``collect_pending_migrations()``.
            platform_options: Platform-specific render-time options forwarded to the auto runner.

        """
        self.config = config
        self.target_platform = target_platform
        if self.target_platform != "azure_databricks":
            raise NotImplementedError(f"Target platform {self.target_platform} is not supported yet.")
        self._executor: SQLExecutor | None = connector.get_sql_executor() if connector else None
        self._state_client: StateClient = state_client or LocalStateClient()
        self._platform_options = platform_options

    def _get_sql_file_state(self, sql_folder: str) -> DatabaseState:
        """Load the SQL file state from the specified folder.

        Args:
            sql_folder: Path to folder containing SQL files.

        Returns:
            DatabaseState built from parsed SQL definitions.

        """
        if self.target_platform != "azure_databricks":
            raise NotImplementedError(f"Target platform {self.target_platform} is not supported yet.")
        mapping = process_sql(sql_folder, recursive=True)
        return DatabaseState(databases=mapping.to_databases().databases)

    def _get_database_state(self, state_path: str) -> DatabaseState:
        """Load the database state from the specified path."""
        return self._state_client.read_state(state_path, DatabaseState)

    def compare_sql_file_state_with_database_statefile(self):
        """Compare the SQL file state with the database state and determine which migrations need to be run."""
        # This is where you would implement logic to compare the current database state
        # with the desired state and determine which migrations need to be run.
        sql_file_state = self._get_sql_file_state(self.config.database_sql_folder)
        database_state = self._get_database_state(self.config.database_statefile_path)
        migrations = MigrationGenerator.generate_migrations(
            current_state=database_state,
            desired_state=sql_file_state,
            platform_options=self._platform_options,
        )
        MigrationGenerator.print_migrations(migrations)

    def collect_pending_migrations(self, current_database_state: DatabaseState) -> list[tuple[str, list[Migration]]]:
        """Return pending migrations across all phases in deployment order without executing them.

        Args:
            current_database_state: Live database state fetched from the target.

        Returns:
            List of ``(phase_name, migrations)`` pairs for pre-deployment, auto,
            and post-deployment phases. Migrations within each phase are in dependency order.

        """
        result: list[tuple[str, list[Migration]]] = []

        # Pre-deployment (file-based)
        pre_runner = FileMigrationRunner(state_client=self._state_client)
        pre_pending = pre_runner.list_pending_migrations(
            statefile_path=self.config.pre_deployment_migration_state_path,
            root_path=self.config.migration_root_path,
            glob_pattern=self.config.pre_deployment_glob,
            conflict_strategy=self.config.conflict_strategy,
        )
        result.append(("pre-deployment", pre_pending))

        # Auto (diff-based)
        desired_state = self._get_sql_file_state(self.config.database_sql_folder)
        auto_runner = AutoMigrationRunner(
            state_client=self._state_client,
            platform_options=self._platform_options,
        )
        auto_pending = auto_runner.list_pending_migrations(
            current_state=current_database_state,
            desired_state=desired_state,
        )
        result.append(("deployment", auto_pending))

        # Post-deployment (file-based)
        post_runner = FileMigrationRunner(state_client=self._state_client)
        post_pending = post_runner.list_pending_migrations(
            statefile_path=self.config.post_deployment_migration_state_path,
            root_path=self.config.migration_root_path,
            glob_pattern=self.config.post_deployment_glob,
            conflict_strategy=self.config.conflict_strategy,
        )
        result.append(("post-deployment", post_pending))

        return result

    def deploy(self, current_database_state: DatabaseState) -> None:
        """Orchestrate the migration process: pre-deployment → auto → post-deployment.

        Args:
            current_database_state: Live database state fetched from the target.

        """
        # Pre-deployment
        pre_runner = FileMigrationRunner(executor=self._executor, state_client=self._state_client)
        pre_runner.execute_migrations(
            statefile_path=self.config.pre_deployment_migration_state_path,
            root_path=self.config.migration_root_path,
            glob_pattern=self.config.pre_deployment_glob,
            conflict_strategy=self.config.conflict_strategy,
        )

        # Auto
        desired_state = self._get_sql_file_state(self.config.database_sql_folder)
        auto_runner = AutoMigrationRunner(
            executor=self._executor,
            state_client=self._state_client,
            platform_options=self._platform_options,
        )
        auto_runner.execute_migrations(
            current_state=current_database_state,
            desired_state=desired_state,
            statefile_path=self.config.auto_migration_state_path,
        )

        # Post-deployment
        post_runner = FileMigrationRunner(executor=self._executor, state_client=self._state_client)
        post_runner.execute_migrations(
            statefile_path=self.config.post_deployment_migration_state_path,
            root_path=self.config.migration_root_path,
            glob_pattern=self.config.post_deployment_glob,
            conflict_strategy=self.config.conflict_strategy,
        )

__init__(config, target_platform='azure_databricks', connector=None, state_client=None, platform_options=None)

Initialize the MigrationOrchestrator.

Parameters:

Name	Type	Description	Default
config	MigrationOrchestratorConfig	Migration paths and state file configuration.	required
target_platform	Literal['azure_databricks']	Target platform for migrations (for now, only "azure_databricks" is supported).	'azure_databricks'
state_client	StateClient | None	Client for reading/writing state. Defaults to LocalStateClient.	None
connector	DatabricksConnector | None	Databricks connector. Required to call deploy(); may be omitted when only calling collect_pending_migrations().	None
platform_options	AzureDatabricksOptions | None	Platform-specific render-time options forwarded to the auto runner.	None

Source code in src/cloe_delta_table_manager/migrations/migration_orchestrator.py

def __init__(
    self,
    config: MigrationOrchestratorConfig,
    target_platform: Literal["azure_databricks"] = "azure_databricks",
    connector: DatabricksConnector | None = None,
    state_client: StateClient | None = None,
    platform_options: AzureDatabricksOptions | None = None,
):
    """Initialize the MigrationOrchestrator.

    Args:
        config: Migration paths and state file configuration.
        target_platform: Target platform for migrations (for now, only "azure_databricks" is supported).
        state_client: Client for reading/writing state. Defaults to LocalStateClient.
        connector: Databricks connector. Required to call ``deploy()``; may be omitted
            when only calling ``collect_pending_migrations()``.
        platform_options: Platform-specific render-time options forwarded to the auto runner.

    """
    self.config = config
    self.target_platform = target_platform
    if self.target_platform != "azure_databricks":
        raise NotImplementedError(f"Target platform {self.target_platform} is not supported yet.")
    self._executor: SQLExecutor | None = connector.get_sql_executor() if connector else None
    self._state_client: StateClient = state_client or LocalStateClient()
    self._platform_options = platform_options

collect_pending_migrations(current_database_state)

Return pending migrations across all phases in deployment order without executing them.

Parameters:

Name	Type	Description	Default
current_database_state	DatabaseState	Live database state fetched from the target.	required

Returns:

Type	Description
list[tuple[str, list[Migration]]]	List of (phase_name, migrations) pairs for pre-deployment, auto,
list[tuple[str, list[Migration]]]	and post-deployment phases. Migrations within each phase are in dependency order.

Source code in src/cloe_delta_table_manager/migrations/migration_orchestrator.py

def collect_pending_migrations(self, current_database_state: DatabaseState) -> list[tuple[str, list[Migration]]]:
    """Return pending migrations across all phases in deployment order without executing them.

    Args:
        current_database_state: Live database state fetched from the target.

    Returns:
        List of ``(phase_name, migrations)`` pairs for pre-deployment, auto,
        and post-deployment phases. Migrations within each phase are in dependency order.

    """
    result: list[tuple[str, list[Migration]]] = []

    # Pre-deployment (file-based)
    pre_runner = FileMigrationRunner(state_client=self._state_client)
    pre_pending = pre_runner.list_pending_migrations(
        statefile_path=self.config.pre_deployment_migration_state_path,
        root_path=self.config.migration_root_path,
        glob_pattern=self.config.pre_deployment_glob,
        conflict_strategy=self.config.conflict_strategy,
    )
    result.append(("pre-deployment", pre_pending))

    # Auto (diff-based)
    desired_state = self._get_sql_file_state(self.config.database_sql_folder)
    auto_runner = AutoMigrationRunner(
        state_client=self._state_client,
        platform_options=self._platform_options,
    )
    auto_pending = auto_runner.list_pending_migrations(
        current_state=current_database_state,
        desired_state=desired_state,
    )
    result.append(("deployment", auto_pending))

    # Post-deployment (file-based)
    post_runner = FileMigrationRunner(state_client=self._state_client)
    post_pending = post_runner.list_pending_migrations(
        statefile_path=self.config.post_deployment_migration_state_path,
        root_path=self.config.migration_root_path,
        glob_pattern=self.config.post_deployment_glob,
        conflict_strategy=self.config.conflict_strategy,
    )
    result.append(("post-deployment", post_pending))

    return result

compare_sql_file_state_with_database_statefile()

Compare the SQL file state with the database state and determine which migrations need to be run.

Source code in src/cloe_delta_table_manager/migrations/migration_orchestrator.py

def compare_sql_file_state_with_database_statefile(self):
    """Compare the SQL file state with the database state and determine which migrations need to be run."""
    # This is where you would implement logic to compare the current database state
    # with the desired state and determine which migrations need to be run.
    sql_file_state = self._get_sql_file_state(self.config.database_sql_folder)
    database_state = self._get_database_state(self.config.database_statefile_path)
    migrations = MigrationGenerator.generate_migrations(
        current_state=database_state,
        desired_state=sql_file_state,
        platform_options=self._platform_options,
    )
    MigrationGenerator.print_migrations(migrations)

deploy(current_database_state)

Orchestrate the migration process: pre-deployment → auto → post-deployment.

Parameters:

Name	Type	Description	Default
current_database_state	DatabaseState	Live database state fetched from the target.	required

Source code in src/cloe_delta_table_manager/migrations/migration_orchestrator.py

def deploy(self, current_database_state: DatabaseState) -> None:
    """Orchestrate the migration process: pre-deployment → auto → post-deployment.

    Args:
        current_database_state: Live database state fetched from the target.

    """
    # Pre-deployment
    pre_runner = FileMigrationRunner(executor=self._executor, state_client=self._state_client)
    pre_runner.execute_migrations(
        statefile_path=self.config.pre_deployment_migration_state_path,
        root_path=self.config.migration_root_path,
        glob_pattern=self.config.pre_deployment_glob,
        conflict_strategy=self.config.conflict_strategy,
    )

    # Auto
    desired_state = self._get_sql_file_state(self.config.database_sql_folder)
    auto_runner = AutoMigrationRunner(
        executor=self._executor,
        state_client=self._state_client,
        platform_options=self._platform_options,
    )
    auto_runner.execute_migrations(
        current_state=current_database_state,
        desired_state=desired_state,
        statefile_path=self.config.auto_migration_state_path,
    )

    # Post-deployment
    post_runner = FileMigrationRunner(executor=self._executor, state_client=self._state_client)
    post_runner.execute_migrations(
        statefile_path=self.config.post_deployment_migration_state_path,
        root_path=self.config.migration_root_path,
        glob_pattern=self.config.post_deployment_glob,
        conflict_strategy=self.config.conflict_strategy,
    )

MigrationOrchestratorConfig

Bases: BaseModel

Configure migration paths and state file locations for the orchestrator.

Source code in src/cloe_delta_table_manager/migrations/migration_orchestrator.py

class MigrationOrchestratorConfig(BaseModel):
    """Configure migration paths and state file locations for the orchestrator."""

    model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")

    database_sql_folder: str
    migration_root_path: str
    pre_deployment_glob: str
    post_deployment_glob: str

    conflict_strategy: MigrationConflictStrategy = RaiseOnConflict()

    # Set by ProjectConfig.prepare_for_target from the target's state_folder.
    database_statefile_path: str = ""
    solution_statefile_path: str = ""
    pre_deployment_migration_state_path: str = ""
    post_deployment_migration_state_path: str = ""
    auto_migration_state_path: str = ""

    @field_validator("conflict_strategy", mode="before")
    @classmethod
    def _coerce_conflict_strategy(cls, v: object) -> MigrationConflictStrategy:
        """Accept a strategy name string in place of a strategy instance.

        Args:
            v: A ``MigrationConflictStrategy`` instance or one of the strings
                ``"raise"``, ``"warn"``, or ``"ignore"``.

        Returns:
            The resolved ``MigrationConflictStrategy`` instance.

        Raises:
            ValueError: If *v* is a string that does not match a known strategy.

        """
        if isinstance(v, MigrationConflictStrategy):
            return v
        _strategies: dict[str, MigrationConflictStrategy] = {
            "raise": RaiseOnConflict(),
            "warn": WarnOnConflict(),
            "ignore": IgnoreConflict(),
        }
        if isinstance(v, str):
            try:
                return _strategies[v]
            except KeyError as e:
                raise ValueError(f"Unknown conflict strategy {v!r}. Valid options are: {list(_strategies)}") from e
        raise ValueError(f"Expected a MigrationConflictStrategy instance or a string, got {type(v)!r}")