Skip to content

Index

Delta Table Manager package for managing delta table migrations and state.

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 
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
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 
33
34
35
36
37
38
39
40
41
42
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 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
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

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

FileMigrationRunner

Bases: MigrationRunner

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

Source code in src/cloe_delta_table_manager/migrations/runner/file.py 
 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
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
225
226
227
228
229
230
231
232
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 
24
25
26
27
28
29
30
31
32
33
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 
35
36
37
38
39
40
41
42
43
44
45
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 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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()

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.

    """

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}")

MigrationRunner

Bases: ABC

Manages reading, ordering, and executing database migrations.

Source code in src/cloe_delta_table_manager/migrations/runner/base.py 
 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
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 
24
25
26
27
28
29
30
31
32
33
34
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 
86
87
88
@abstractmethod
def execute_migrations(self, *args, **kwargs) -> None:
    """Execute migrations."""

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"

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,
        )