Skip to content

graflo.architecture.evolution.structure

Structure-plane evolution ops: introducing and retargeting graph entities.

The contract vocabulary could remove, merge, rename and project, but not create. Everything here closes that half of the loop so a change set can describe a growing graph without falling back to a binary compose.

apply_add_edges(manifest, op)

Add new logical edge relations between existing vertex types.

Source code in graflo/architecture/evolution/structure.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def apply_add_edges(manifest: GraphManifest, op: AddEdgesOp) -> None:
    """Add new logical edge relations between existing vertex types."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("add_edges requires graph_schema")

    edge_config = schema.core_schema.edge_config
    known_vertices = schema.core_schema.vertex_config.vertex_set
    existing_ids = {edge.edge_id for edge in edge_config.edges}

    collisions = sorted(
        str((edge.source, edge.target, edge.relation))
        for edge in op.edges
        if (edge.source, edge.target, edge.relation) in existing_ids
    )
    if collisions:
        raise ValueError(f"add_edges: edges already exist: {collisions}")

    unknown = sorted(
        {
            name
            for edge in op.edges
            for name in (edge.source, edge.target)
            if name not in known_vertices
        }
    )
    if unknown:
        raise ValueError(
            f"add_edges: unknown endpoint vertex types: {unknown}; add them with "
            "AddVerticesOp first"
        )

    edge_config.edges = [
        *edge_config.edges,
        *(edge.model_copy(deep=True) for edge in op.edges),
    ]
    schema.finish_init()

apply_add_vertices(manifest, op)

Add new logical vertex types to the schema.

Source code in graflo/architecture/evolution/structure.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def apply_add_vertices(manifest: GraphManifest, op: AddVerticesOp) -> None:
    """Add new logical vertex types to the schema."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("add_vertices requires graph_schema")

    vertex_config = schema.core_schema.vertex_config
    existing = vertex_config.vertex_set
    collisions = sorted(
        vertex.name for vertex in op.vertices if vertex.name in existing
    )
    if collisions:
        raise ValueError(
            f"add_vertices: vertices already exist: {collisions}; use "
            "RenameVerticesOp or MergeVerticesOp instead"
        )

    vertex_config.vertices = [
        *vertex_config.vertices,
        *(vertex.model_copy(deep=True) for vertex in op.vertices),
    ]
    schema.finish_init()

apply_retarget_edges(manifest, op)

Repoint edges at different endpoint vertex types, keeping everything else.

Mutates manifest in place: rewrites the edge's source / target, moves the matching db_profile physical specs onto the new EdgeId, and repoints pipeline edge steps that named the old triple.

Source code in graflo/architecture/evolution/structure.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def apply_retarget_edges(manifest: GraphManifest, op: RetargetEdgesOp) -> None:
    """Repoint edges at different endpoint vertex types, keeping everything else.

    Mutates *manifest* in place: rewrites the edge's ``source`` / ``target``, moves the
    matching ``db_profile`` physical specs onto the new ``EdgeId``, and repoints
    pipeline edge steps that named the old triple.
    """
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("retarget_edges requires graph_schema")

    edge_config = schema.core_schema.edge_config
    by_edge_id = {edge.edge_id: edge for edge in edge_config.edges}
    known_vertices = schema.core_schema.vertex_config.vertex_set

    unknown_edges = sorted(
        str(entry.edge_id()) for entry in op.edges if entry.edge_id() not in by_edge_id
    )
    if unknown_edges:
        raise ValueError(f"retarget_edges: unknown edges: {unknown_edges}")

    unknown_vertices = sorted(
        {
            name
            for entry in op.edges
            for name in (entry.new_source, entry.new_target)
            if name is not None and name not in known_vertices
        }
    )
    if unknown_vertices:
        raise ValueError(
            f"retarget_edges: unknown endpoint vertex types: {unknown_vertices}"
        )

    collisions = sorted(
        str(entry.retargeted_edge_id())
        for entry in op.edges
        if entry.retargeted_edge_id() in by_edge_id
    )
    if collisions:
        raise ValueError(
            f"retarget_edges: retargeted edges collide with existing ones: "
            f"{collisions}; use MergeEdgesOp to combine them"
        )

    mapping: dict[EdgeId, tuple[str, str]] = {}
    for entry in op.edges:
        edge = by_edge_id[entry.edge_id()]
        new_source, new_target, _ = entry.retargeted_edge_id()
        mapping[entry.edge_id()] = (new_source, new_target)

        for spec in schema.db_profile.edge_specs:
            if (spec.source, spec.target, spec.relation) == entry.edge_id():
                spec.source = new_source
                spec.target = new_target

        edge.source = new_source
        edge.target = new_target
        logger.debug(
            "retarget_edges: %s -> %s", entry.edge_id(), entry.retargeted_edge_id()
        )

    # ``EdgeConfig._edges_map`` is a private index keyed on EdgeId and rebuilt by a
    # model validator. Mutating an edge's endpoints in place leaves it keyed on the
    # pre-retarget id, which the Schema-level spec check then reads as a dangling
    # reference. Reassigning the list re-runs the validator and rebuilds the index.
    edge_config.edges = list(edge_config.edges)

    schema.db_profile = _revalidated_profile(schema)
    schema.finish_init()

    if manifest.ingestion_model is not None:
        from .apply import _rebuild_ingestion_with_pipeline_rewrite
        from .rewrite import rewrite_edge_endpoints_in_pipeline

        _rebuild_ingestion_with_pipeline_rewrite(
            manifest,
            lambda pipeline: rewrite_edge_endpoints_in_pipeline(pipeline, mapping),
        )