Skip to content

graflo.migrate.models

Typed migration domain models.

MigrationOperation

Bases: BaseModel

A typed migration operation emitted from schema diff.

Source code in graflo/migrate/models.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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."
    )

MigrationPlan

Bases: BaseModel

Ordered migration operations ready for preview or execution.

Source code in graflo/migrate/models.py 
77
78
79
80
81
82
83
84
85
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

MigrationRecord

Bases: BaseModel

Applied migration metadata persisted by migration store.

Source code in graflo/migrate/models.py 
88
89
90
91
92
93
94
95
96
97
98
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(timezone.utc).isoformat()
    )

OperationType

Bases: StrEnum

Canonical migration operation type identifiers.

Source code in graflo/migrate/models.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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"
    REKEY_VERTEX = "REKEY_VERTEX"
    CHANGE_INDEX = "CHANGE_INDEX"

RiskLevel

Bases: StrEnum

Risk levels attached to migration operations.

Source code in graflo/migrate/models.py 
12
13
14
15
16
17
18
class RiskLevel(StrEnum):
    """Risk levels attached to migration operations."""

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

SchemaConflict

Bases: BaseModel

A conflict discovered during planning or validation.

Source code in graflo/migrate/models.py 
61
62
63
64
65
66
class SchemaConflict(BaseModel):
    """A conflict discovered during planning or validation."""

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

SchemaDiffResult

Bases: BaseModel

Structured schema diff with operations and diagnostics.

Source code in graflo/migrate/models.py 
69
70
71
72
73
74
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)