graflo.migrate

Migration planning and execution primitives.

Modules:

Name	Description
diff	Schema diff engine for migration planning.
drift	Live schema drift: what a database holds that its declared schema does not.
emitters	Backend emitters for migration execution.
executor	Migration execution orchestrator.
io	I/O utilities for migration workflows.
models	Typed migration domain models.
planner	Migration plan ordering and policy filters.
risk	Risk classification and compatibility helpers.
store	Migration history persistence.

Attributes

__all__ = ['LiveSchemaDrift', 'MigrationOperation', 'MigrationPlan', 'MigrationPlanner', 'MigrationRecord', 'OperationType', 'RiskLevel', 'SchemaConflict', 'SchemaDiff', 'SchemaDiffResult', 'compare_live_schema'] module-attribute

Classes

LiveSchemaDrift

Bases: BaseModel

Presence-only differences between a live database and its schema.

Names are logical where the declared schema maps them, and the raw storage name where it does not (anything undeclared has no logical name).

Source code in graflo/migrate/drift.py

class LiveSchemaDrift(BaseModel):
    """Presence-only differences between a live database and its schema.

    Names are logical where the declared schema maps them, and the raw storage
    name where it does not (anything undeclared has no logical name).
    """

    undeclared_vertices: list[str] = Field(
        default_factory=list,
        description="Vertex types in the database that the schema does not declare.",
    )
    missing_vertices: list[str] = Field(
        default_factory=list,
        description="Declared vertex types with no node in the database (or sample).",
    )
    undeclared_properties: dict[str, list[str]] = Field(
        default_factory=dict,
        description="Per declared vertex type, properties present but not declared.",
    )
    missing_properties: dict[str, list[str]] = Field(
        default_factory=dict,
        description=(
            "Per declared vertex type present in the database, declared "
            "properties no examined node carries."
        ),
    )
    undeclared_edges: list[EdgeKey] = Field(
        default_factory=list,
        description="(source, relation, target) patterns in the data, not declared.",
    )
    missing_edges: list[EdgeKey] = Field(
        default_factory=list,
        description="Declared (source, relation, target) patterns with no edge.",
    )
    sampled: bool = Field(
        default=True,
        description=(
            "True when introspection examined a sample of rows per type, so a "
            "missing entry means 'not seen in the sample', not 'absent'."
        ),
    )

    @computed_field
    @property
    def has_drift(self) -> bool:
        """True when the database holds anything the schema does not declare."""
        return bool(
            self.undeclared_vertices
            or self.undeclared_properties
            or self.undeclared_edges
        )

Attributes

has_drift property

True when the database holds anything the schema does not declare.

missing_edges = Field(default_factory=list, description='Declared (source, relation, target) patterns with no edge.') class-attribute instance-attribute

missing_properties = Field(default_factory=dict, description='Per declared vertex type present in the database, declared properties no examined node carries.') class-attribute instance-attribute

missing_vertices = Field(default_factory=list, description='Declared vertex types with no node in the database (or sample).') class-attribute instance-attribute

sampled = Field(default=True, description="True when introspection examined a sample of rows per type, so a missing entry means 'not seen in the sample', not 'absent'.") class-attribute instance-attribute

undeclared_edges = Field(default_factory=list, description='(source, relation, target) patterns in the data, not declared.') class-attribute instance-attribute

undeclared_properties = Field(default_factory=dict, description='Per declared vertex type, properties present but not declared.') class-attribute instance-attribute

undeclared_vertices = Field(default_factory=list, description='Vertex types in the database that the schema does not declare.') class-attribute instance-attribute

MigrationOperation

Bases: BaseModel

A typed migration operation emitted from schema diff.

Source code in graflo/migrate/models.py

class MigrationOperation(BaseModel):
    """A typed migration operation emitted from schema diff."""

    op_type: OperationType = Field(..., description="Operation type identifier.")
    target: str = Field(..., description="Operation target path.")
    old_value: Any = Field(default=None, description="Old value, when relevant.")
    new_value: Any = Field(default=None, description="New value, when relevant.")
    risk: RiskLevel = Field(..., description="Operation risk level.")
    reversible: bool = Field(
        default=True,
        description="Whether operation is reversible without external backup.",
    )
    metadata: dict[str, Any] = Field(
        default_factory=dict, description="Additional operation metadata."
    )

Attributes

metadata = Field(default_factory=dict, description='Additional operation metadata.') class-attribute instance-attribute

new_value = Field(default=None, description='New value, when relevant.') class-attribute instance-attribute

old_value = Field(default=None, description='Old value, when relevant.') class-attribute instance-attribute

op_type = Field(..., description='Operation type identifier.') class-attribute instance-attribute

reversible = Field(default=True, description='Whether operation is reversible without external backup.') class-attribute instance-attribute

risk = Field(..., description='Operation risk level.') class-attribute instance-attribute

target = Field(..., description='Operation target path.') class-attribute instance-attribute

MigrationPlan

Bases: BaseModel

Ordered migration operations ready for preview or execution.

Source code in graflo/migrate/models.py

class MigrationPlan(BaseModel):
    """Ordered migration operations ready for preview or execution."""

    operations: list[MigrationOperation] = Field(default_factory=list)
    blocked_operations: list[MigrationOperation] = Field(default_factory=list)
    warnings: list[str] = Field(default_factory=list)

    def is_empty(self) -> bool:
        return len(self.operations) == 0 and len(self.blocked_operations) == 0

Attributes

blocked_operations = Field(default_factory=list) class-attribute instance-attribute

operations = Field(default_factory=list) class-attribute instance-attribute

warnings = Field(default_factory=list) class-attribute instance-attribute

Methods:

is_empty()

Source code in graflo/migrate/models.py

def is_empty(self) -> bool:
    return len(self.operations) == 0 and len(self.blocked_operations) == 0

MigrationPlanner

Translate schema diff result into ordered execution plan.

Source code in graflo/migrate/planner.py

class MigrationPlanner:
    """Translate schema diff result into ordered execution plan."""

    def __init__(self, allow_high_risk: bool = False):
        self.allow_high_risk = allow_high_risk

    def build(self, diff_result: SchemaDiffResult) -> MigrationPlan:
        """Build ordered plan with risk gate filtering."""
        ordered_ops = sorted(
            diff_result.operations,
            key=lambda op: (OP_ORDER.get(op.op_type, 9999), op.target),
        )
        runnable: list[MigrationOperation] = []
        blocked: list[MigrationOperation] = []

        for op in ordered_ops:
            if self.allow_high_risk or is_low_risk(op):
                runnable.append(op)
            else:
                blocked.append(op)

        warnings = list(diff_result.warnings)
        if blocked and not self.allow_high_risk:
            warnings.append(
                "High-risk operations are blocked by default. Re-run with explicit allow flag in future guarded workflow."
            )
        return MigrationPlan(
            operations=runnable,
            blocked_operations=blocked,
            warnings=warnings,
        )

Attributes

allow_high_risk = allow_high_risk instance-attribute

Methods:

__init__(allow_high_risk=False)

Source code in graflo/migrate/planner.py

def __init__(self, allow_high_risk: bool = False):
    self.allow_high_risk = allow_high_risk

build(diff_result)

Build ordered plan with risk gate filtering.

Source code in graflo/migrate/planner.py

def build(self, diff_result: SchemaDiffResult) -> MigrationPlan:
    """Build ordered plan with risk gate filtering."""
    ordered_ops = sorted(
        diff_result.operations,
        key=lambda op: (OP_ORDER.get(op.op_type, 9999), op.target),
    )
    runnable: list[MigrationOperation] = []
    blocked: list[MigrationOperation] = []

    for op in ordered_ops:
        if self.allow_high_risk or is_low_risk(op):
            runnable.append(op)
        else:
            blocked.append(op)

    warnings = list(diff_result.warnings)
    if blocked and not self.allow_high_risk:
        warnings.append(
            "High-risk operations are blocked by default. Re-run with explicit allow flag in future guarded workflow."
        )
    return MigrationPlan(
        operations=runnable,
        blocked_operations=blocked,
        warnings=warnings,
    )

MigrationRecord

Bases: BaseModel

Applied migration metadata persisted by migration store.

Source code in graflo/migrate/models.py

class MigrationRecord(BaseModel):
    """Applied migration metadata persisted by migration store."""

    revision: str
    schema_hash: str
    backend: str
    operations: list[MigrationOperation | OperationType] = Field(
        default_factory=list,
        description=(
            "The operations applied, in order. Records written before full "
            "operations were stored hold only each operation's type."
        ),
    )
    reversible: bool = True
    applied_at: str = Field(default_factory=lambda: datetime.now(UTC).isoformat())

Attributes

applied_at = Field(default_factory=lambda: datetime.now(UTC).isoformat()) class-attribute instance-attribute

backend instance-attribute

operations = Field(default_factory=list, description="The operations applied, in order. Records written before full operations were stored hold only each operation's type.") class-attribute instance-attribute

reversible = True class-attribute instance-attribute

revision instance-attribute

schema_hash instance-attribute

OperationType

Bases: StrEnum

Canonical migration operation type identifiers.

Source code in graflo/migrate/models.py

class OperationType(StrEnum):
    """Canonical migration operation type identifiers."""

    ADD_VERTEX = "ADD_VERTEX"
    ADD_EDGE = "ADD_EDGE"
    ADD_VERTEX_FIELD = "ADD_VERTEX_FIELD"
    ADD_EDGE_FIELD = "ADD_EDGE_FIELD"
    ADD_VERTEX_INDEX = "ADD_VERTEX_INDEX"
    ADD_EDGE_INDEX = "ADD_EDGE_INDEX"
    CHANGE_VERTEX_FIELD_TYPE = "CHANGE_VERTEX_FIELD_TYPE"
    CHANGE_EDGE_FIELD_TYPE = "CHANGE_EDGE_FIELD_TYPE"
    REMOVE_EDGE_INDEX = "REMOVE_EDGE_INDEX"
    REMOVE_VERTEX_INDEX = "REMOVE_VERTEX_INDEX"
    REMOVE_EDGE_FIELD = "REMOVE_EDGE_FIELD"
    REMOVE_VERTEX_FIELD = "REMOVE_VERTEX_FIELD"
    REMOVE_EDGE = "REMOVE_EDGE"
    REMOVE_VERTEX = "REMOVE_VERTEX"
    CHANGE_EDGE_IDENTITY = "CHANGE_EDGE_IDENTITY"
    CHANGE_VERTEX_IDENTITY = "CHANGE_VERTEX_IDENTITY"
    CHANGE_SECONDARY_IDENTITY = "CHANGE_SECONDARY_IDENTITY"
    REKEY_VERTEX = "REKEY_VERTEX"
    CHANGE_INDEX = "CHANGE_INDEX"

Attributes

ADD_EDGE = 'ADD_EDGE' class-attribute instance-attribute

ADD_EDGE_FIELD = 'ADD_EDGE_FIELD' class-attribute instance-attribute

ADD_EDGE_INDEX = 'ADD_EDGE_INDEX' class-attribute instance-attribute

ADD_VERTEX = 'ADD_VERTEX' class-attribute instance-attribute

ADD_VERTEX_FIELD = 'ADD_VERTEX_FIELD' class-attribute instance-attribute

ADD_VERTEX_INDEX = 'ADD_VERTEX_INDEX' class-attribute instance-attribute

CHANGE_EDGE_FIELD_TYPE = 'CHANGE_EDGE_FIELD_TYPE' class-attribute instance-attribute

CHANGE_EDGE_IDENTITY = 'CHANGE_EDGE_IDENTITY' class-attribute instance-attribute

CHANGE_INDEX = 'CHANGE_INDEX' class-attribute instance-attribute

CHANGE_SECONDARY_IDENTITY = 'CHANGE_SECONDARY_IDENTITY' class-attribute instance-attribute

CHANGE_VERTEX_FIELD_TYPE = 'CHANGE_VERTEX_FIELD_TYPE' class-attribute instance-attribute

CHANGE_VERTEX_IDENTITY = 'CHANGE_VERTEX_IDENTITY' class-attribute instance-attribute

REKEY_VERTEX = 'REKEY_VERTEX' class-attribute instance-attribute

REMOVE_EDGE = 'REMOVE_EDGE' class-attribute instance-attribute

REMOVE_EDGE_FIELD = 'REMOVE_EDGE_FIELD' class-attribute instance-attribute

REMOVE_EDGE_INDEX = 'REMOVE_EDGE_INDEX' class-attribute instance-attribute

REMOVE_VERTEX = 'REMOVE_VERTEX' class-attribute instance-attribute

REMOVE_VERTEX_FIELD = 'REMOVE_VERTEX_FIELD' class-attribute instance-attribute

REMOVE_VERTEX_INDEX = 'REMOVE_VERTEX_INDEX' class-attribute instance-attribute

RiskLevel

Bases: StrEnum

Risk levels attached to migration operations.

Source code in graflo/migrate/models.py

class RiskLevel(StrEnum):
    """Risk levels attached to migration operations."""

    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"
    CRITICAL = "CRITICAL"

Attributes

CRITICAL = 'CRITICAL' class-attribute instance-attribute

HIGH = 'HIGH' class-attribute instance-attribute

LOW = 'LOW' class-attribute instance-attribute

MEDIUM = 'MEDIUM' class-attribute instance-attribute

SchemaConflict

Bases: BaseModel

A conflict discovered during planning or validation.

Source code in graflo/migrate/models.py

class SchemaConflict(BaseModel):
    """A conflict discovered during planning or validation."""

    key: str
    message: str
    risk: RiskLevel = RiskLevel.MEDIUM

Attributes

key instance-attribute

message instance-attribute

risk = RiskLevel.MEDIUM class-attribute instance-attribute

SchemaDiff

Compute a typed structural diff between two schemas.

Source code in graflo/migrate/diff.py

class SchemaDiff:
    """Compute a typed structural diff between two schemas."""

    def __init__(self, schema_old: Schema, schema_new: Schema):
        self.schema_old = schema_old
        self.schema_new = schema_new
        self.schema_old.finish_init()
        self.schema_new.finish_init()
        self._result: SchemaDiffResult | None = None

    def compare(self) -> SchemaDiffResult:
        """Build a full diff result including operations/conflicts/warnings."""
        operations: list[MigrationOperation] = []
        conflicts: list[SchemaConflict] = []
        warnings: list[str] = []

        operations.extend(self._diff_vertices(conflicts))
        operations.extend(self._diff_edges(conflicts))
        operations.extend(self._diff_database_features())

        self._result = SchemaDiffResult(
            operations=operations, conflicts=conflicts, warnings=warnings
        )
        return self._result

    def operations(self) -> list[MigrationOperation]:
        """Return diff operations, calculating if needed."""
        if self._result is None:
            self.compare()
        if self._result is None:
            return []
        return self._result.operations

    def is_backward_compatible(self) -> bool:
        """True when schema_new is additive compared to schema_old."""
        return is_backward_compatible_operations(self.operations())

    def risk_assessment(self) -> dict[str, str]:
        """Map operation keys to risk labels."""
        risk_map: dict[str, str] = {}
        for op in self.operations():
            risk_map[f"{op.op_type}:{op.target}"] = op.risk.value
        return risk_map

    def conflicts(self) -> list[SchemaConflict]:
        """Return conflicts from latest compare call."""
        if self._result is None:
            self.compare()
        if self._result is None:
            return []
        return self._result.conflicts

    def _diff_vertices(
        self, conflicts: list[SchemaConflict]
    ) -> list[MigrationOperation]:
        old_vertices = {
            vertex.name: vertex
            for vertex in self.schema_old.core_schema.vertex_config.vertices
        }
        new_vertices = {
            vertex.name: vertex
            for vertex in self.schema_new.core_schema.vertex_config.vertices
        }
        old_names = set(old_vertices)
        new_names = set(new_vertices)
        operations: list[MigrationOperation] = []

        for name in sorted(new_names - old_names):
            operations.append(
                self._op(
                    OperationType.ADD_VERTEX,
                    f"vertex:{name}",
                    None,
                    new_vertices[name].to_dict(),
                )
            )
        for name in sorted(old_names - new_names):
            operations.append(
                self._op(
                    OperationType.REMOVE_VERTEX,
                    f"vertex:{name}",
                    old_vertices[name].to_dict(),
                    None,
                    reversible=False,
                )
            )

        for name in sorted(old_names & new_names):
            old_vertex = old_vertices[name]
            new_vertex = new_vertices[name]

            operations.extend(
                self._diff_vertex_identity(name, old_vertex, new_vertex, conflicts)
            )

            old_fields = _field_map(old_vertex.properties)
            new_fields = _field_map(new_vertex.properties)
            old_field_names = set(old_fields)
            new_field_names = set(new_fields)

            for field_name in sorted(new_field_names - old_field_names):
                operations.append(
                    self._op(
                        OperationType.ADD_VERTEX_FIELD,
                        f"vertex:{name}:field:{field_name}",
                        None,
                        {"name": field_name, **new_fields[field_name]},
                    )
                )
            for field_name in sorted(old_field_names - new_field_names):
                operations.append(
                    self._op(
                        OperationType.REMOVE_VERTEX_FIELD,
                        f"vertex:{name}:field:{field_name}",
                        {"name": field_name, **old_fields[field_name]},
                        None,
                        reversible=False,
                    )
                )
            for field_name in sorted(old_field_names & new_field_names):
                if old_fields[field_name] != new_fields[field_name]:
                    operations.append(
                        self._op(
                            OperationType.CHANGE_VERTEX_FIELD_TYPE,
                            f"vertex:{name}:field:{field_name}:type",
                            old_fields[field_name],
                            new_fields[field_name],
                            reversible=False,
                        )
                    )

        return operations

    @staticmethod
    def _identity_state(vertex: Vertex) -> dict[str, Any]:
        """Everything that decides how a vertex is keyed at write time.

        Comparing only ``identity`` misses a mode change: a vertex moving from a
        natural key to a hash keeps ``identity == ["id"]`` on both sides while its
        write-time key semantics change completely.
        """
        return {
            "mode": vertex.identity_mode,
            "identity": list(vertex.identity),
            "hash_identity_properties": list(vertex.hash_identity_properties),
            # A funnel resolves to mode ``hash`` with no flat properties, so the
            # branches themselves are the only thing that distinguishes two
            # funnel policies — or a funnel from a flat hash.
            "identity_funnel": (
                vertex.identity_funnel.to_minimal_canonical_dict()
                if vertex.identity_funnel is not None
                else None
            ),
        }

    @staticmethod
    def _secondary_identity_state(vertex: Vertex) -> list[dict[str, Any]]:
        return sorted(
            (
                {"name": entry.name, "fields": sorted(entry.fields)}
                for entry in vertex.secondary_identities
            ),
            key=lambda entry: (entry["name"] or "", tuple(entry["fields"])),
        )

    @staticmethod
    def _requires_rekey(old_state: dict[str, Any], new_state: dict[str, Any]) -> bool:
        """Whether stored vertex keys become invalid under the new identity.

        A mode change always invalidates them — the key is computed differently.
        A natural-to-natural change only invalidates them when the new key is not
        derivable from the old one; widening a composite key (adding a field) leaves
        every existing key still addressable.
        """
        if old_state["mode"] != new_state["mode"]:
            return True
        if new_state["mode"] == "hash":
            # Covers flat<->funnel and funnel<->funnel: any change to the digest
            # inputs, the branch order or the branch ids yields different keys.
            return (
                old_state["hash_identity_properties"]
                != new_state["hash_identity_properties"]
                or old_state["identity_funnel"] != new_state["identity_funnel"]
            )
        if new_state["mode"] != "natural":
            return False
        return not set(old_state["identity"]).issubset(set(new_state["identity"]))

    def _diff_vertex_identity(
        self,
        name: str,
        old_vertex: Vertex,
        new_vertex: Vertex,
        conflicts: list[SchemaConflict],
    ) -> list[MigrationOperation]:
        """Identity, identity mode, and secondary identities for one vertex."""
        operations: list[MigrationOperation] = []

        old_state = self._identity_state(old_vertex)
        new_state = self._identity_state(new_vertex)
        if old_state != new_state:
            operations.append(
                self._op(
                    OperationType.CHANGE_VERTEX_IDENTITY,
                    f"vertex:{name}:identity",
                    old_state,
                    new_state,
                    reversible=False,
                )
            )
            conflicts.append(
                SchemaConflict(
                    key=f"vertex:{name}:identity",
                    message="Vertex identity changed; requires explicit rekey strategy.",
                    risk=classify_operation(OperationType.CHANGE_VERTEX_IDENTITY),
                )
            )
            if self._requires_rekey(old_state, new_state):
                operations.append(
                    self._op(
                        OperationType.REKEY_VERTEX,
                        f"vertex:{name}:rekey",
                        old_state,
                        new_state,
                        reversible=False,
                    )
                )
                conflicts.append(
                    SchemaConflict(
                        key=f"vertex:{name}:rekey",
                        message=(
                            "Stored vertex keys are no longer derivable from the new "
                            "identity; existing vertices must be re-keyed."
                        ),
                        risk=classify_operation(OperationType.REKEY_VERTEX),
                    )
                )

        old_secondary = self._secondary_identity_state(old_vertex)
        new_secondary = self._secondary_identity_state(new_vertex)
        if old_secondary != new_secondary:
            operations.append(
                self._op(
                    OperationType.CHANGE_SECONDARY_IDENTITY,
                    f"vertex:{name}:secondary_identities",
                    old_secondary,
                    new_secondary,
                )
            )

        return operations

    def _diff_edges(self, conflicts: list[SchemaConflict]) -> list[MigrationOperation]:
        old_edges = {
            edge.edge_id: edge for edge in self.schema_old.core_schema.edge_config.edges
        }
        new_edges = {
            edge.edge_id: edge for edge in self.schema_new.core_schema.edge_config.edges
        }
        old_ids = set(old_edges)
        new_ids = set(new_edges)
        operations: list[MigrationOperation] = []

        for edge_id in sorted(new_ids - old_ids):
            edge = new_edges[edge_id]
            operations.append(
                self._op(
                    OperationType.ADD_EDGE, f"edge:{edge_id}", None, edge.to_dict()
                )
            )
        for edge_id in sorted(old_ids - new_ids):
            edge = old_edges[edge_id]
            operations.append(
                self._op(
                    OperationType.REMOVE_EDGE,
                    f"edge:{edge_id}",
                    edge.to_dict(),
                    None,
                    reversible=False,
                )
            )

        for edge_id in sorted(old_ids & new_ids):
            old_edge = old_edges[edge_id]
            new_edge = new_edges[edge_id]

            if old_edge.identities != new_edge.identities:
                operations.append(
                    self._op(
                        OperationType.CHANGE_EDGE_IDENTITY,
                        f"edge:{edge_id}:identity",
                        old_edge.identities,
                        new_edge.identities,
                        reversible=False,
                    )
                )
                conflicts.append(
                    SchemaConflict(
                        key=f"edge:{edge_id}:identity",
                        message="Edge identity changed; may impact deduplication semantics.",
                        risk=classify_operation(OperationType.CHANGE_EDGE_IDENTITY),
                    )
                )

            old_direct = _field_map(old_edge.properties)
            new_direct = _field_map(new_edge.properties)
            old_names = set(old_direct)
            new_names = set(new_direct)

            for field_name in sorted(new_names - old_names):
                operations.append(
                    self._op(
                        OperationType.ADD_EDGE_FIELD,
                        f"edge:{edge_id}:field:{field_name}",
                        None,
                        {"name": field_name, **new_direct[field_name]},
                    )
                )
            for field_name in sorted(old_names - new_names):
                operations.append(
                    self._op(
                        OperationType.REMOVE_EDGE_FIELD,
                        f"edge:{edge_id}:field:{field_name}",
                        {"name": field_name, **old_direct[field_name]},
                        None,
                        reversible=False,
                    )
                )
            for field_name in sorted(old_names & new_names):
                if old_direct[field_name] != new_direct[field_name]:
                    operations.append(
                        self._op(
                            OperationType.CHANGE_EDGE_FIELD_TYPE,
                            f"edge:{edge_id}:field:{field_name}:type",
                            old_direct[field_name],
                            new_direct[field_name],
                            reversible=False,
                        )
                    )

        return operations

    def _diff_database_features(self) -> list[MigrationOperation]:
        operations: list[MigrationOperation] = []
        all_vertices = (
            self.schema_old.core_schema.vertex_config.vertex_set
            | self.schema_new.core_schema.vertex_config.vertex_set
        )

        for vertex_name in sorted(all_vertices):
            old_ix = (
                _vertex_index_tuples(self.schema_old, vertex_name)
                if vertex_name in self.schema_old.core_schema.vertex_config.vertex_set
                else set()
            )
            new_ix = (
                _vertex_index_tuples(self.schema_new, vertex_name)
                if vertex_name in self.schema_new.core_schema.vertex_config.vertex_set
                else set()
            )
            for ix in sorted(new_ix - old_ix):
                operations.append(
                    self._op(
                        OperationType.ADD_VERTEX_INDEX,
                        f"vertex:{vertex_name}:index:{ix}",
                        None,
                        ix,
                    )
                )
            for ix in sorted(old_ix - new_ix):
                operations.append(
                    self._op(
                        OperationType.REMOVE_VERTEX_INDEX,
                        f"vertex:{vertex_name}:index:{ix}",
                        ix,
                        None,
                    )
                )

        old_edges = {
            edge.edge_id: edge for edge in self.schema_old.core_schema.edge_config.edges
        }
        new_edges = {
            edge.edge_id: edge for edge in self.schema_new.core_schema.edge_config.edges
        }
        all_edge_ids = set(old_edges) | set(new_edges)
        for edge_id in sorted(all_edge_ids):
            old_ix = (
                _edge_index_tuples(self.schema_old, old_edges[edge_id])
                if edge_id in old_edges
                else set()
            )
            new_ix = (
                _edge_index_tuples(self.schema_new, new_edges[edge_id])
                if edge_id in new_edges
                else set()
            )
            for ix in sorted(new_ix - old_ix):
                operations.append(
                    self._op(
                        OperationType.ADD_EDGE_INDEX,
                        f"edge:{edge_id}:index:{ix}",
                        None,
                        ix,
                    )
                )
            for ix in sorted(old_ix - new_ix):
                operations.append(
                    self._op(
                        OperationType.REMOVE_EDGE_INDEX,
                        f"edge:{edge_id}:index:{ix}",
                        ix,
                        None,
                    )
                )

        return operations

    @staticmethod
    def _op(
        op_type: OperationType,
        target: str,
        old_value: Any,
        new_value: Any,
        reversible: bool = True,
    ) -> MigrationOperation:
        return MigrationOperation(
            op_type=op_type,
            target=target,
            old_value=old_value,
            new_value=new_value,
            risk=classify_operation(op_type),
            reversible=reversible,
        )

Attributes

schema_new = schema_new instance-attribute

schema_old = schema_old instance-attribute

Methods:

__init__(schema_old, schema_new)

Source code in graflo/migrate/diff.py

def __init__(self, schema_old: Schema, schema_new: Schema):
    self.schema_old = schema_old
    self.schema_new = schema_new
    self.schema_old.finish_init()
    self.schema_new.finish_init()
    self._result: SchemaDiffResult | None = None

compare()

Build a full diff result including operations/conflicts/warnings.

Source code in graflo/migrate/diff.py

def compare(self) -> SchemaDiffResult:
    """Build a full diff result including operations/conflicts/warnings."""
    operations: list[MigrationOperation] = []
    conflicts: list[SchemaConflict] = []
    warnings: list[str] = []

    operations.extend(self._diff_vertices(conflicts))
    operations.extend(self._diff_edges(conflicts))
    operations.extend(self._diff_database_features())

    self._result = SchemaDiffResult(
        operations=operations, conflicts=conflicts, warnings=warnings
    )
    return self._result

conflicts()

Return conflicts from latest compare call.

Source code in graflo/migrate/diff.py

def conflicts(self) -> list[SchemaConflict]:
    """Return conflicts from latest compare call."""
    if self._result is None:
        self.compare()
    if self._result is None:
        return []
    return self._result.conflicts

is_backward_compatible()

True when schema_new is additive compared to schema_old.

Source code in graflo/migrate/diff.py

def is_backward_compatible(self) -> bool:
    """True when schema_new is additive compared to schema_old."""
    return is_backward_compatible_operations(self.operations())

operations()

Return diff operations, calculating if needed.

Source code in graflo/migrate/diff.py

def operations(self) -> list[MigrationOperation]:
    """Return diff operations, calculating if needed."""
    if self._result is None:
        self.compare()
    if self._result is None:
        return []
    return self._result.operations

risk_assessment()

Map operation keys to risk labels.

Source code in graflo/migrate/diff.py

def risk_assessment(self) -> dict[str, str]:
    """Map operation keys to risk labels."""
    risk_map: dict[str, str] = {}
    for op in self.operations():
        risk_map[f"{op.op_type}:{op.target}"] = op.risk.value
    return risk_map

SchemaDiffResult

Bases: BaseModel

Structured schema diff with operations and diagnostics.

Source code in graflo/migrate/models.py

class SchemaDiffResult(BaseModel):
    """Structured schema diff with operations and diagnostics."""

    operations: list[MigrationOperation] = Field(default_factory=list)
    conflicts: list[SchemaConflict] = Field(default_factory=list)
    warnings: list[str] = Field(default_factory=list)

Attributes

conflicts = Field(default_factory=list) class-attribute instance-attribute

operations = Field(default_factory=list) class-attribute instance-attribute

warnings = Field(default_factory=list) class-attribute instance-attribute

Functions:

compare_live_schema(declared, observed, *, db_flavor, sampled=True)

Compare an introspected schema with the declared one, by presence only.

Parameters:

Name	Type	Description	Default
declared	Schema	The schema the database is supposed to follow.	required
observed	Schema	What Connection.introspect_graph_schema recovered. Its vertex and relation names are storage names.	required
db_flavor	DBType	The database's flavor, for mapping logical names to storage names.	required
sampled	bool	Whether introspection sampled rows rather than reading a complete catalogue.	True

Returns:

Name	Type	Description
LiveSchemaDrift	LiveSchemaDrift	every difference, with lists sorted for stable output.

Source code in graflo/migrate/drift.py

def compare_live_schema(
    declared: Schema,
    observed: Schema,
    *,
    db_flavor: DBType,
    sampled: bool = True,
) -> LiveSchemaDrift:
    """Compare an introspected schema with the declared one, by presence only.

    Args:
        declared: The schema the database is supposed to follow.
        observed: What ``Connection.introspect_graph_schema`` recovered. Its
            vertex and relation names are storage names.
        db_flavor: The database's flavor, for mapping logical names to storage
            names.
        sampled: Whether introspection sampled rows rather than reading a
            complete catalogue.

    Returns:
        LiveSchemaDrift: every difference, with lists sorted for stable output.
    """
    # The names the database was deployed with: those the profile records,
    # plus the ones the write path fills in for names the flavor cannot store.
    declared = with_physical_names(declared, db_flavor)
    db_aware = declared.resolve_db_aware(db_flavor)
    declared_vertices = declared.core_schema.vertex_config
    profile = declared.db_profile
    logical_of = {
        db_aware.vertex_config.vertex_dbname(name): name
        for name in declared_vertices.vertex_set
    }

    def logical(storage_name: str) -> str:
        return logical_of.get(storage_name, storage_name)

    def logical_properties(vertex: str, stored: list[str]) -> set[str]:
        back = {
            physical: name
            for name, physical in profile.vertex_property_map(vertex).items()
        }
        return {back.get(prop, prop) for prop in stored}

    observed_props: dict[str, set[str]] = {
        logical(v.name): logical_properties(logical(v.name), v.property_names)
        for v in observed.core_schema.vertex_config.vertices
    }

    drift = LiveSchemaDrift(sampled=sampled)
    drift.undeclared_vertices = sorted(
        name for name in observed_props if name not in declared_vertices.vertex_set
    )
    drift.missing_vertices = sorted(
        name for name in declared_vertices.vertex_set if name not in observed_props
    )
    for name in sorted(declared_vertices.vertex_set):
        present = observed_props.get(name)
        if present is None:
            continue
        declared_names = set(declared_vertices[name].property_names) | set(
            db_aware.vertex_config.identity_fields(name)
        )
        extra = sorted(present - declared_names)
        absent = sorted(set(declared_vertices[name].property_names) - present)
        if extra:
            drift.undeclared_properties[name] = extra
        if absent:
            drift.missing_properties[name] = absent

    declared_edges: set[EdgeKey] = {
        (
            edge.source,
            _edge_storage_name(db_aware, edge, db_flavor),
            edge.target,
        )
        for edge in declared.core_schema.edge_config.edges
    }
    observed_edges: set[EdgeKey] = set()
    # A backend that stores edges by collection may report no relation name;
    # such an edge is matched on its endpoints alone, to every declared edge
    # between them, rather than reported as drift it cannot be shown to be.
    matched_by_endpoints: set[EdgeKey] = set()
    for edge in observed.core_schema.edge_config.edges:
        source, target = logical(edge.source), logical(edge.target)
        if edge.relation:
            observed_edges.add((source, edge.relation, target))
            continue
        candidates = {
            key for key in declared_edges if (key[0], key[2]) == (source, target)
        }
        if candidates:
            matched_by_endpoints |= candidates
        else:
            observed_edges.add((source, "", target))
    drift.undeclared_edges = sorted(observed_edges - declared_edges)
    drift.missing_edges = sorted(declared_edges - observed_edges - matched_by_endpoints)
    return drift