Skip to content

migration_graph_service

Service for managing migration dependency graphs.

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