Skip to content

graflo.migrate.planner

Migration plan ordering and policy filters.

Attributes

OP_ORDER = {OperationType.ADD_VERTEX: 10, OperationType.ADD_EDGE: 20, OperationType.ADD_VERTEX_FIELD: 30, OperationType.ADD_EDGE_FIELD: 40, OperationType.ADD_VERTEX_INDEX: 50, OperationType.ADD_EDGE_INDEX: 60, OperationType.CHANGE_VERTEX_FIELD_TYPE: 70, OperationType.CHANGE_EDGE_FIELD_TYPE: 80, OperationType.REMOVE_EDGE_INDEX: 90, OperationType.REMOVE_VERTEX_INDEX: 100, OperationType.CHANGE_SECONDARY_IDENTITY: 105, OperationType.REMOVE_EDGE_FIELD: 110, OperationType.REMOVE_VERTEX_FIELD: 120, OperationType.REMOVE_EDGE: 130, OperationType.REMOVE_VERTEX: 140, OperationType.CHANGE_EDGE_IDENTITY: 150, OperationType.CHANGE_VERTEX_IDENTITY: 160, OperationType.REKEY_VERTEX: 170} module-attribute

Classes

MigrationPlanner

Translate schema diff result into ordered execution plan.

Source code in graflo/migrate/planner.py 
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
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 
38
39
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 
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
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,
    )

Functions: