Skip to content

graflo.migrate.planner

Migration plan ordering and policy filters.

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,
        )

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,
    )