Skip to content

Index

Migrations module for managing database migrations.

FileMigration

Bases: Migration

Represents a user-defined migration script with additional metadata.

Source code in src/cloe_delta_table_manager/migrations/migration.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class FileMigration(Migration):
    """Represents a user-defined migration script with additional metadata."""

    depends_on_paths: list[Path] = Field(default_factory=list)
    source_path: Path

    @field_validator("depends_on_paths", mode="before")
    @classmethod
    def validate_depends_on_path(cls, v):
        """Validate that depends_on_paths contains only relative paths."""
        paths = [Path(item) if isinstance(item, str) else item for item in v]
        for path in paths:
            if path.is_absolute():
                raise ValueError(f"Absolute paths are not allowed in depends_on_paths: {path}")
        return paths

    @model_validator(mode="after")
    def resolve_depends_on_paths(self):
        """Resolve depends_on_paths relative to source_path directory."""
        if self.depends_on_paths and self.source_path:
            source_dir = Path(self.source_path).parent
            self.depends_on_paths = [source_dir / path for path in self.depends_on_paths]
        return self

resolve_depends_on_paths()

Resolve depends_on_paths relative to source_path directory.

Source code in src/cloe_delta_table_manager/migrations/migration.py 
120
121
122
123
124
125
126
@model_validator(mode="after")
def resolve_depends_on_paths(self):
    """Resolve depends_on_paths relative to source_path directory."""
    if self.depends_on_paths and self.source_path:
        source_dir = Path(self.source_path).parent
        self.depends_on_paths = [source_dir / path for path in self.depends_on_paths]
    return self

validate_depends_on_path(v) classmethod

Validate that depends_on_paths contains only relative paths.

Source code in src/cloe_delta_table_manager/migrations/migration.py 
110
111
112
113
114
115
116
117
118
@field_validator("depends_on_paths", mode="before")
@classmethod
def validate_depends_on_path(cls, v):
    """Validate that depends_on_paths contains only relative paths."""
    paths = [Path(item) if isinstance(item, str) else item for item in v]
    for path in paths:
        if path.is_absolute():
            raise ValueError(f"Absolute paths are not allowed in depends_on_paths: {path}")
    return paths

IgnoreConflict

Bases: MigrationConflictStrategy

Silently ignores modified or removed migrations and continues execution.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
74
75
76
77
78
79
80
81
class IgnoreConflict(MigrationConflictStrategy):
    """Silently ignores modified or removed migrations and continues execution."""

    def on_modified(self, migrations: list[Migration]) -> None:
        """Do nothing."""

    def on_removed(self, migrations: list[Migration]) -> None:
        """Do nothing."""

on_modified(migrations)

Do nothing.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
77
78
def on_modified(self, migrations: list[Migration]) -> None:
    """Do nothing."""

on_removed(migrations)

Do nothing.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
80
81
def on_removed(self, migrations: list[Migration]) -> None:
    """Do nothing."""

Migration

Bases: BaseModel

Represents a database migration with operations and dependencies.

Source code in src/cloe_delta_table_manager/migrations/migration.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class Migration(BaseModel):
    """Represents a database migration with operations and dependencies."""

    operations: list[Operation]
    status: MigrationStatus = MigrationStatus.NEW
    depends_on: list[Self] = Field(default_factory=list)

    @computed_field
    @property
    def operation_hash(self) -> int:
        """Compute a deterministic migration id as a SHA-256 hash of all operations.

        Uses hashlib instead of Python's built-in hash() to guarantee identical
        values across processes regardless of PYTHONHASHSEED.
        """
        if not self.operations:
            return 0
        content = "|".join(f"{op.index}:{op.operation_type.value}:{';'.join(op.commands)}" for op in self.operations)
        digest = hashlib.sha256(content.encode()).digest()
        return int.from_bytes(digest[:8], "big")

    @classmethod
    def from_diff(
        cls,
        diff: Diff,
        on_unsupported: OnUnsupported = "fail",
        platform_options: AzureDatabricksOptions | None = None,
    ) -> Migration:
        """Create a Migration object from a typed Diff object."""
        from cloe_delta_table_manager.diff import (  # noqa: PLC0415
            DatabricksStatementGenerator,
        )

        operation = Operation(
            commands=DatabricksStatementGenerator(
                on_unsupported=on_unsupported,
                options=platform_options,
            ).generate(diff),
            operation_type=_operation_type_for(diff),
            index=0,
        )
        return cls(operations=[operation], depends_on=[])

    def __str__(self) -> str:
        """Migration string representation showing id and status."""
        return f"Migration(id={self.operation_hash}, status={self.status})"

    def __repr__(self) -> str:
        """Detailed string representation showing id, status, number of operations, and dependencies."""
        return f"Migration(id={self.operation_hash}, status={self.status}, operations={len(self.operations)}, depends_on={len(self.depends_on)})"

operation_hash property

Compute a deterministic migration id as a SHA-256 hash of all operations.

Uses hashlib instead of Python's built-in hash() to guarantee identical values across processes regardless of PYTHONHASHSEED.

__repr__()

Detailed string representation showing id, status, number of operations, and dependencies.

Source code in src/cloe_delta_table_manager/migrations/migration.py 
75
76
77
def __repr__(self) -> str:
    """Detailed string representation showing id, status, number of operations, and dependencies."""
    return f"Migration(id={self.operation_hash}, status={self.status}, operations={len(self.operations)}, depends_on={len(self.depends_on)})"

__str__()

Migration string representation showing id and status.

Source code in src/cloe_delta_table_manager/migrations/migration.py 
71
72
73
def __str__(self) -> str:
    """Migration string representation showing id and status."""
    return f"Migration(id={self.operation_hash}, status={self.status})"

from_diff(diff, on_unsupported='fail', platform_options=None) classmethod

Create a Migration object from a typed Diff object.

Source code in src/cloe_delta_table_manager/migrations/migration.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@classmethod
def from_diff(
    cls,
    diff: Diff,
    on_unsupported: OnUnsupported = "fail",
    platform_options: AzureDatabricksOptions | None = None,
) -> Migration:
    """Create a Migration object from a typed Diff object."""
    from cloe_delta_table_manager.diff import (  # noqa: PLC0415
        DatabricksStatementGenerator,
    )

    operation = Operation(
        commands=DatabricksStatementGenerator(
            on_unsupported=on_unsupported,
            options=platform_options,
        ).generate(diff),
        operation_type=_operation_type_for(diff),
        index=0,
    )
    return cls(operations=[operation], depends_on=[])

MigrationConflictStrategy

Bases: ABC

Abstract base class defining how to react to modified or removed migrations.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class MigrationConflictStrategy(ABC):
    """Abstract base class defining how to react to modified or removed migrations."""

    @abstractmethod
    def on_modified(self, migrations: list[Migration]) -> None:
        """Handle migrations that have been modified since the last execution.

        Args:
            migrations: Modified migrations detected by the runner.

        """

    @abstractmethod
    def on_removed(self, migrations: list[Migration]) -> None:
        """Handle migrations that have been removed since the last execution.

        Args:
            migrations: Removed migrations detected by the runner.

        """

on_modified(migrations) abstractmethod

Handle migrations that have been modified since the last execution.

Parameters:

Name Type Description Default
migrations list[Migration]

Modified migrations detected by the runner.

 required
Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
14
15
16
17
18
19
20
21
@abstractmethod
def on_modified(self, migrations: list[Migration]) -> None:
    """Handle migrations that have been modified since the last execution.

    Args:
        migrations: Modified migrations detected by the runner.

    """

on_removed(migrations) abstractmethod

Handle migrations that have been removed since the last execution.

Parameters:

Name Type Description Default
migrations list[Migration]

Removed migrations detected by the runner.

 required
Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
23
24
25
26
27
28
29
30
@abstractmethod
def on_removed(self, migrations: list[Migration]) -> None:
    """Handle migrations that have been removed since the last execution.

    Args:
        migrations: Removed migrations detected by the runner.

    """

MigrationGraphService

Manages a directed acyclic graph (DAG) of migrations and their dependencies.

This service builds and maintains a graph where nodes represent migrations and edges represent dependencies between them. It can identify which migrations are ready to execute based on their dependencies' states.

The graph is lazily built and automatically rebuilt when the migrations list changes, so you never need to manually refresh it.

Attributes:

Name Type Description
migrations list[Migration]

List of migrations to manage.
Source code in src/cloe_delta_table_manager/migrations/migration_graph_service.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class MigrationGraphService:
    """Manages a directed acyclic graph (DAG) of migrations and their dependencies.

    This service builds and maintains a graph where nodes represent migrations and
    edges represent dependencies between them. It can identify which migrations are
    ready to execute based on their dependencies' states.

    The graph is lazily built and automatically rebuilt when the migrations list
    changes, so you never need to manually refresh it.

    Attributes:
        migrations: List of migrations to manage.

    """

    def __init__(self, migrations: list[Migration]) -> None:
        """Initialize the MigrationGraphService with a list of migrations.

        Args:
            migrations: List of Migration objects to manage.

        """
        self.migrations: list[Migration] = migrations
        self._cached_graph: nx.DiGraph | None = None
        self._cached_migration_ids: set[int] | None = None
        self._validate_acyclic()

    @property
    def graph(self) -> nx.DiGraph:
        """Get the migration dependency graph.

        The graph is lazily built and automatically rebuilt if the migrations
        list has changed since the last access.

        Returns:
            A directed graph where nodes are migration IDs and edges
            represent dependencies.

        """
        current_ids = {m.operation_hash for m in self.migrations}

        # Check if we need to rebuild the graph
        if self._cached_graph is None or self._cached_migration_ids != current_ids:
            self._cached_migration_ids = current_ids
            self._cached_graph = self._create_graph()

        return self._cached_graph

    def _create_graph(self) -> nx.DiGraph:
        """Create a directed acyclic graph (DAG) of migrations and their dependencies.

        Each node in the graph represents a migration (identified by its operation_hash),
        and each edge represents a dependency relationship where the source node must
        be executed before the target node.

        Returns:
            A NetworkX DiGraph with operation hashes as nodes.

        """
        g: nx.DiGraph = nx.DiGraph()

        # Add all migrations as nodes using their deterministic operation_hash
        migration_hashes = {m.operation_hash for m in self.migrations}
        g.add_nodes_from(migration_hashes)

        # Add edges for dependencies
        edges = set()
        for migration in self.migrations:
            dep: Migration
            for dep in migration.depends_on:  # type: ignore[assignment]
                # Edge from dependency to migration (dependency must run first)
                edges.add((dep.operation_hash, migration.operation_hash))

        g.add_edges_from(edges)

        return g

    def _validate_acyclic(self) -> None:
        """Validate that the migration graph is acyclic.

        Raises:
            RuntimeError: If a circular dependency is detected in the migration graph.

        """
        # Access graph property to ensure it's up to date
        graph = self.graph

        if not nx.is_directed_acyclic_graph(graph):
            cycles = list(nx.simple_cycles(graph))
            # Convert cycles from IDs to readable format
            migration_map = {m.operation_hash: m for m in self.migrations}
            readable_cycles = []
            for cycle in cycles:
                cycle_names = []
                for node_id in cycle:
                    migration = migration_map[node_id]
                    # Use source_path if available, otherwise use id
                    name = (
                        str(migration.source_path)
                        if hasattr(migration, "source_path") and migration.source_path
                        else f"Migration(id={node_id})"
                    )
                    cycle_names.append(name)
                readable_cycles.append(" -> ".join(cycle_names))

            cycle_str = "\n".join(f"  - {cycle}" for cycle in readable_cycles)
            raise RuntimeError(f"Circular dependency detected in migrations. Cycles found:\n{cycle_str}")

    def get_execution_order(self) -> list[Migration]:
        """Return NEW migrations in topological execution order.

        Returns:
            List of NEW migrations ordered so that all dependencies come before dependants.

        """
        migration_map = {m.operation_hash: m for m in self.migrations}
        new_hashes = {m.operation_hash for m in self.migrations if m.status == MigrationStatus.NEW}
        return [migration_map[h] for h in nx.topological_sort(self.graph) if h in new_hashes]

    def get_ready_migration(self) -> Migration | None:
        """Get the next migration that is ready to be applied.

        A migration is ready to execute if:
        1. Its state is NEW
        2. All its dependencies have state APPLIED

        Returns:
            The first Migration object that is ready to execute, or None if no migrations are ready.

        """
        for migration in self.migrations:
            if migration.status != MigrationStatus.NEW:
                continue

            # Check if all dependencies are applied
            all_dependencies_applied = all(
                dep.status == MigrationStatus.APPLIED  # type: ignore[attr-defined]
                for dep in migration.depends_on
            )

            if all_dependencies_applied:
                migration.status = MigrationStatus.PENDING
                return migration

        return None

graph property

Get the migration dependency graph.

The graph is lazily built and automatically rebuilt if the migrations list has changed since the last access.

Returns:

Type Description
DiGraph

A directed graph where nodes are migration IDs and edges
DiGraph

represent dependencies.

__init__(migrations)

Initialize the MigrationGraphService with a list of migrations.

Parameters:

Name Type Description Default
migrations list[Migration]

List of Migration objects to manage.

 required
Source code in src/cloe_delta_table_manager/migrations/migration_graph_service.py 
23
24
25
26
27
28
29
30
31
32
33
def __init__(self, migrations: list[Migration]) -> None:
    """Initialize the MigrationGraphService with a list of migrations.

    Args:
        migrations: List of Migration objects to manage.

    """
    self.migrations: list[Migration] = migrations
    self._cached_graph: nx.DiGraph | None = None
    self._cached_migration_ids: set[int] | None = None
    self._validate_acyclic()

get_execution_order()

Return NEW migrations in topological execution order.

Returns:

Type Description
list[Migration]

List of NEW migrations ordered so that all dependencies come before dependants.
Source code in src/cloe_delta_table_manager/migrations/migration_graph_service.py 
116
117
118
119
120
121
122
123
124
125
def get_execution_order(self) -> list[Migration]:
    """Return NEW migrations in topological execution order.

    Returns:
        List of NEW migrations ordered so that all dependencies come before dependants.

    """
    migration_map = {m.operation_hash: m for m in self.migrations}
    new_hashes = {m.operation_hash for m in self.migrations if m.status == MigrationStatus.NEW}
    return [migration_map[h] for h in nx.topological_sort(self.graph) if h in new_hashes]

get_ready_migration()

Get the next migration that is ready to be applied.

A migration is ready to execute if: 1. Its state is NEW 2. All its dependencies have state APPLIED

Returns:

Type Description
Migration | None

The first Migration object that is ready to execute, or None if no migrations are ready.
Source code in src/cloe_delta_table_manager/migrations/migration_graph_service.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def get_ready_migration(self) -> Migration | None:
    """Get the next migration that is ready to be applied.

    A migration is ready to execute if:
    1. Its state is NEW
    2. All its dependencies have state APPLIED

    Returns:
        The first Migration object that is ready to execute, or None if no migrations are ready.

    """
    for migration in self.migrations:
        if migration.status != MigrationStatus.NEW:
            continue

        # Check if all dependencies are applied
        all_dependencies_applied = all(
            dep.status == MigrationStatus.APPLIED  # type: ignore[attr-defined]
            for dep in migration.depends_on
        )

        if all_dependencies_applied:
            migration.status = MigrationStatus.PENDING
            return migration

    return None

MigrationOrchestrator

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

Source code in src/cloe_delta_table_manager/migrations/migration_orchestrator.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
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 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
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 
128
129
130
131
132
133
134
135
136
137
138
139
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 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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}")

MigrationStatus

Bases: Enum

Enum representing the status of a migration.

Source code in src/cloe_delta_table_manager/migrations/migration.py 
18
19
20
21
22
23
24
25
class MigrationStatus(Enum):
    """Enum representing the status of a migration."""

    UNDEFINED = "undefined"
    NEW = "new"
    PENDING = "pending"
    APPLIED = "applied"
    FAILED = "failed"

Operation

Bases: BaseModel

Represents a single operation in a migration.

Source code in src/cloe_delta_table_manager/migrations/operation.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class Operation(BaseModel):
    """Represents a single operation in a migration."""

    model_config = ConfigDict(frozen=True)

    commands: list[str]
    operation_type: OperationType
    index: int

    def __str__(self) -> str:
        """Human-readable representation showing operation type and commands."""
        return f"[{self.index}] {self.operation_type.value}: {'; '.join(self.commands)}"

    def __hash__(self) -> int:
        """Make Operation hashable by converting list to tuple."""
        return hash((tuple(self.commands), self.operation_type, self.index))

__hash__()

Make Operation hashable by converting list to tuple.

Source code in src/cloe_delta_table_manager/migrations/operation.py 
30
31
32
def __hash__(self) -> int:
    """Make Operation hashable by converting list to tuple."""
    return hash((tuple(self.commands), self.operation_type, self.index))

__str__()

Human-readable representation showing operation type and commands.

Source code in src/cloe_delta_table_manager/migrations/operation.py 
26
27
28
def __str__(self) -> str:
    """Human-readable representation showing operation type and commands."""
    return f"[{self.index}] {self.operation_type.value}: {'; '.join(self.commands)}"

OperationType

Bases: StrEnum

Enum for operation types.

Source code in src/cloe_delta_table_manager/migrations/operation.py 
 8
 9
10
11
12
13
14
class OperationType(StrEnum):
    """Enum for operation types."""

    CREATE = "create"
    DROP = "drop"
    ALTER = "alter"
    USER = "user"

RaiseOnConflict

Bases: MigrationConflictStrategy

Raises ValueError when modified or removed migrations are detected.

This is the default behaviour, preserving strict integrity guarantees.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class RaiseOnConflict(MigrationConflictStrategy):
    """Raises ValueError when modified or removed migrations are detected.

    This is the default behaviour, preserving strict integrity guarantees.
    """

    def on_modified(self, migrations: list[Migration]) -> None:
        """Raise ValueError listing all modified migrations."""
        if migrations:
            migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
            raise ValueError(f"The following migrations have been modified since the last execution:\n{migration_str}")

    def on_removed(self, migrations: list[Migration]) -> None:
        """Raise ValueError listing all removed migrations."""
        if migrations:
            migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
            raise ValueError(f"The following migrations have been removed since the last execution:\n{migration_str}")

on_modified(migrations)

Raise ValueError listing all modified migrations.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
39
40
41
42
43
def on_modified(self, migrations: list[Migration]) -> None:
    """Raise ValueError listing all modified migrations."""
    if migrations:
        migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
        raise ValueError(f"The following migrations have been modified since the last execution:\n{migration_str}")

on_removed(migrations)

Raise ValueError listing all removed migrations.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
45
46
47
48
49
def on_removed(self, migrations: list[Migration]) -> None:
    """Raise ValueError listing all removed migrations."""
    if migrations:
        migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
        raise ValueError(f"The following migrations have been removed since the last execution:\n{migration_str}")

WarnOnConflict

Bases: MigrationConflictStrategy

Emits a :mod:warnings warning and continues when conflicts are detected.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class WarnOnConflict(MigrationConflictStrategy):
    """Emits a :mod:`warnings` warning and continues when conflicts are detected."""

    def on_modified(self, migrations: list[Migration]) -> None:
        """Emit a warning listing all modified migrations."""
        if migrations:
            migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
            warnings.warn(
                f"The following migrations have been modified since the last execution:\n{migration_str}",
                stacklevel=3,
            )

    def on_removed(self, migrations: list[Migration]) -> None:
        """Emit a warning listing all removed migrations."""
        if migrations:
            migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
            warnings.warn(
                f"The following migrations have been removed since the last execution:\n{migration_str}",
                stacklevel=3,
            )

on_modified(migrations)

Emit a warning listing all modified migrations.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
55
56
57
58
59
60
61
62
def on_modified(self, migrations: list[Migration]) -> None:
    """Emit a warning listing all modified migrations."""
    if migrations:
        migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
        warnings.warn(
            f"The following migrations have been modified since the last execution:\n{migration_str}",
            stacklevel=3,
        )

on_removed(migrations)

Emit a warning listing all removed migrations.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
64
65
66
67
68
69
70
71
def on_removed(self, migrations: list[Migration]) -> None:
    """Emit a warning listing all removed migrations."""
    if migrations:
        migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
        warnings.warn(
            f"The following migrations have been removed since the last execution:\n{migration_str}",
            stacklevel=3,
        )