Skip to content

graflo.migrate.risk

Risk classification and compatibility helpers.

Attributes

CRITICAL_RISK_OPS = {OperationType.CHANGE_VERTEX_IDENTITY, OperationType.CHANGE_EDGE_IDENTITY, OperationType.REKEY_VERTEX} module-attribute

HIGH_RISK_OPS = {OperationType.REMOVE_VERTEX_FIELD, OperationType.REMOVE_EDGE_FIELD, OperationType.CHANGE_VERTEX_FIELD_TYPE, OperationType.CHANGE_EDGE_FIELD_TYPE, OperationType.REMOVE_VERTEX, OperationType.REMOVE_EDGE} module-attribute

LOW_RISK_OPS = {OperationType.ADD_VERTEX, OperationType.ADD_EDGE, OperationType.ADD_VERTEX_FIELD, OperationType.ADD_EDGE_FIELD, OperationType.ADD_VERTEX_INDEX, OperationType.ADD_EDGE_INDEX} module-attribute

MEDIUM_RISK_OPS = {OperationType.REMOVE_VERTEX_INDEX, OperationType.REMOVE_EDGE_INDEX, OperationType.CHANGE_INDEX, OperationType.CHANGE_SECONDARY_IDENTITY} module-attribute

Classes

Functions:

classify_operation(op_type)

Map operation type to risk level.

Source code in graflo/migrate/risk.py 
41
42
43
44
45
46
47
48
49
50
51
def classify_operation(op_type: OperationType) -> RiskLevel:
    """Map operation type to risk level."""
    if op_type in LOW_RISK_OPS:
        return RiskLevel.LOW
    if op_type in MEDIUM_RISK_OPS:
        return RiskLevel.MEDIUM
    if op_type in HIGH_RISK_OPS:
        return RiskLevel.HIGH
    if op_type in CRITICAL_RISK_OPS:
        return RiskLevel.CRITICAL
    return RiskLevel.MEDIUM

is_backward_compatible_operations(operations)

A migration is backward compatible when all ops are additive low-risk.

Source code in graflo/migrate/risk.py 
59
60
61
def is_backward_compatible_operations(operations: list[MigrationOperation]) -> bool:
    """A migration is backward compatible when all ops are additive low-risk."""
    return all(op.risk == RiskLevel.LOW for op in operations)

is_low_risk(op)

Return True when operation is safe for v1 executor.

Source code in graflo/migrate/risk.py 
54
55
56
def is_low_risk(op: MigrationOperation) -> bool:
    """Return True when operation is safe for v1 executor."""
    return op.risk == RiskLevel.LOW