graflo.migrate.models

Typed migration domain models.

Classes

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

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[str] = Field(default_factory=list)
    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) 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

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