Skip to content

migration_generator

Generates migrations by walking two database states with typed diffs.

MigrationGenerator

Walk state pairs → typed diffs → dependency-wired Migrations.

Source code in src/cloe_delta_table_manager/migrations/migration_generator.py 
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
class MigrationGenerator:
    """Walk state pairs → typed diffs → dependency-wired Migrations."""

    @staticmethod
    def generate_migrations(
        current_state: DatabaseState,
        desired_state: DatabaseState,
        on_unsupported: OnUnsupported = "fail",
        platform_options: AzureDatabricksOptions | None = None,
    ) -> list[Migration]:
        """Generate migrations to transition from current to desired state.

        Args:
            current_state: The state the system is in.
            desired_state: The state the system should be in.
            on_unsupported: ``"fail"`` (default) raises
                :class:`~cloe_delta_table_manager.diff.UnsupportedDiffError` when a
                diff cannot be rendered. ``"skip"`` drops the offending diff
                and all transitively dependent diffs, logging a warning.
            platform_options: Platform-specific render-time options (e.g.
                Databricks ``table_properties``). Forwarded to the statement
                generator.

        """
        diffs = DiffResolver().diff(current_state, desired_state)
        DiffDependencyResolver().resolve(diffs)

        if on_unsupported == "skip":
            diffs = _drop_unsupported_with_dependents(diffs)

        migrations = [
            Migration.from_diff(diff, on_unsupported=on_unsupported, platform_options=platform_options)
            for diff in diffs
        ]
        diff_id_to_migration = {diff.id: migration for diff, migration in zip(diffs, migrations, strict=True)}

        for diff, migration in zip(diffs, migrations, strict=True):
            migration.depends_on = [
                diff_id_to_migration[dep_id] for dep_id in diff.depends_on if dep_id in diff_id_to_migration
            ]

        return migrations

    @staticmethod
    def print_migrations(migrations: list[Migration]) -> None:
        """Print the generated migrations in execution order (dependencies first)."""
        ordered = _topological_sort(migrations)
        print(f"Generated {len(ordered)} migrations:")
        for i, migration in enumerate(ordered):
            print(f"{'#' * 20} {i} {'#' * 20} \n{[op.commands for op in migration.operations]}")

generate_migrations(current_state, desired_state, on_unsupported='fail', platform_options=None) staticmethod

Generate migrations to transition from current to desired state.

Parameters:

Name Type Description Default
current_state DatabaseState

The state the system is in.

 required
desired_state DatabaseState

The state the system should be in.

 required
on_unsupported OnUnsupported

"fail" (default) raises :class:~cloe_delta_table_manager.diff.UnsupportedDiffError when a diff cannot be rendered. "skip" drops the offending diff and all transitively dependent diffs, logging a warning.

 'fail'
platform_options AzureDatabricksOptions | None

Platform-specific render-time options (e.g. Databricks table_properties). Forwarded to the statement generator.

 None
Source code in src/cloe_delta_table_manager/migrations/migration_generator.py 
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
@staticmethod
def generate_migrations(
    current_state: DatabaseState,
    desired_state: DatabaseState,
    on_unsupported: OnUnsupported = "fail",
    platform_options: AzureDatabricksOptions | None = None,
) -> list[Migration]:
    """Generate migrations to transition from current to desired state.

    Args:
        current_state: The state the system is in.
        desired_state: The state the system should be in.
        on_unsupported: ``"fail"`` (default) raises
            :class:`~cloe_delta_table_manager.diff.UnsupportedDiffError` when a
            diff cannot be rendered. ``"skip"`` drops the offending diff
            and all transitively dependent diffs, logging a warning.
        platform_options: Platform-specific render-time options (e.g.
            Databricks ``table_properties``). Forwarded to the statement
            generator.

    """
    diffs = DiffResolver().diff(current_state, desired_state)
    DiffDependencyResolver().resolve(diffs)

    if on_unsupported == "skip":
        diffs = _drop_unsupported_with_dependents(diffs)

    migrations = [
        Migration.from_diff(diff, on_unsupported=on_unsupported, platform_options=platform_options)
        for diff in diffs
    ]
    diff_id_to_migration = {diff.id: migration for diff, migration in zip(diffs, migrations, strict=True)}

    for diff, migration in zip(diffs, migrations, strict=True):
        migration.depends_on = [
            diff_id_to_migration[dep_id] for dep_id in diff.depends_on if dep_id in diff_id_to_migration
        ]

    return migrations

print_migrations(migrations) staticmethod

Print the generated migrations in execution order (dependencies first).

Source code in src/cloe_delta_table_manager/migrations/migration_generator.py 
66
67
68
69
70
71
72
@staticmethod
def print_migrations(migrations: list[Migration]) -> None:
    """Print the generated migrations in execution order (dependencies first)."""
    ordered = _topological_sort(migrations)
    print(f"Generated {len(ordered)} migrations:")
    for i, migration in enumerate(ordered):
        print(f"{'#' * 20} {i} {'#' * 20} \n{[op.commands for op in migration.operations]}")