Skip to content

graflo.architecture.evolution.merge_core

Pure merge helpers for logical vertices and edges.

edge_config_from_edges(edges)

Build a fresh :class:EdgeConfig from a list of edges.

Source code in graflo/architecture/evolution/merge_core.py 
127
128
129
def edge_config_from_edges(edges: list[Edge]) -> EdgeConfig:
    """Build a fresh :class:`EdgeConfig` from a list of edges."""
    return EdgeConfig(edges=edges)

merge_edge_pair(a, b)

Merge two edges with the same :attr:~graflo.architecture.schema.edge.Edge.edge_id.

Source code in graflo/architecture/evolution/merge_core.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def merge_edge_pair(a: Edge, b: Edge) -> Edge:
    """Merge two edges with the same :attr:`~graflo.architecture.schema.edge.Edge.edge_id`."""
    props: dict[str, Field] = {}
    for f in a.properties + b.properties:
        if f.name not in props:
            props[f.name] = f
        else:
            props[f.name] = merge_field_pair(props[f.name], f)

    identities_out: list[list[str]] = []
    seen_rows: set[tuple[str, ...]] = set()
    for row in a.identities + b.identities:
        t = tuple(row)
        if t not in seen_rows:
            seen_rows.add(t)
            identities_out.append(list(row))

    descriptions = [a.description, b.description]
    descriptions = [d for d in descriptions if d]
    desc_out: str | None = None
    if len(descriptions) == 1:
        desc_out = descriptions[0]
    elif len(descriptions) > 1:
        desc_out = " / ".join(descriptions)

    return Edge(
        source=a.source,
        target=a.target,
        relation=a.relation,
        description=desc_out,
        identities=identities_out,
        properties=list(props.values()),
        type=a.type,
        by=a.by,
    )

merge_field_pair(a, b)

Merge two fields with the same name; fail on incompatible types.

Source code in graflo/architecture/evolution/merge_core.py 
10
11
12
13
14
15
16
17
18
def merge_field_pair(a: Field, b: Field) -> Field:
    """Merge two fields with the same name; fail on incompatible types."""
    if a.type is not None and b.type is not None and a.type != b.type:
        raise ValueError(
            f"Cannot merge field {a.name!r}: incompatible types {a.type!r} vs {b.type!r}"
        )
    merged_type = a.type if a.type is not None else b.type
    desc = a.description if a.description else b.description
    return Field(name=a.name, type=merged_type, description=desc)

merge_vertex_models(vertices, into_name)

Union-merge vertex definitions into a single :class:Vertex.

Source code in graflo/architecture/evolution/merge_core.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
def merge_vertex_models(vertices: list[Vertex], into_name: str) -> Vertex:
    """Union-merge vertex definitions into a single :class:`Vertex`."""
    if not vertices:
        raise ValueError("merge_vertex_models requires at least one vertex")

    props: dict[str, Field] = {}
    for v in vertices:
        for f in v.properties:
            if f.name not in props:
                props[f.name] = f
            else:
                props[f.name] = merge_field_pair(props[f.name], f)

    identity_out: list[str] = []
    seen_id: set[str] = set()
    for v in vertices:
        for x in v.identity:
            if x not in seen_id:
                identity_out.append(x)
                seen_id.add(x)

    filters_out: list = []
    for v in vertices:
        filters_out.extend(list(v.filters))

    descriptions = [v.description for v in vertices if v.description]
    if not descriptions:
        desc_out: str | None = None
    elif len(descriptions) == 1:
        desc_out = descriptions[0]
    else:
        desc_out = " / ".join(descriptions)

    return Vertex(
        name=into_name,
        properties=list(props.values()),
        identity=identity_out,
        filters=filters_out,
        description=desc_out,
    )

redirect_and_merge_edges(edges, mapping)

Apply vertex mapping to endpoints, then merge duplicate edge identities.

Source code in graflo/architecture/evolution/merge_core.py 
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
def redirect_and_merge_edges(edges: list[Edge], mapping: dict[str, str]) -> list[Edge]:
    """Apply vertex *mapping* to endpoints, then merge duplicate edge identities."""

    def _map_endpoint(n: str) -> str:
        return mapping.get(n, n)

    redirected: list[Edge] = []
    for e in edges:
        redirected.append(
            e.model_copy(
                update={
                    "source": _map_endpoint(e.source),
                    "target": _map_endpoint(e.target),
                }
            )
        )

    by_id: dict[EdgeId, Edge] = {}
    for e in redirected:
        eid = e.edge_id
        if eid not in by_id:
            by_id[eid] = e
        else:
            by_id[eid] = merge_edge_pair(by_id[eid], e)
    return list(by_id.values())

remap_relation_and_merge_edges(edges, relation_map)

Remap edge relation names and merge duplicate edge identities.

Source code in graflo/architecture/evolution/merge_core.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def remap_relation_and_merge_edges(
    edges: list[Edge], relation_map: dict[str, str]
) -> list[Edge]:
    """Remap edge relation names and merge duplicate edge identities."""
    if not relation_map:
        return list(edges)
    remapped = [
        edge.model_copy(
            update={"relation": relation_map.get(edge.relation, edge.relation)}
        )
        if edge.relation is not None
        else edge
        for edge in edges
    ]
    by_id: dict[EdgeId, Edge] = {}
    for edge in remapped:
        if edge.edge_id in by_id:
            by_id[edge.edge_id] = merge_edge_pair(by_id[edge.edge_id], edge)
            continue
        by_id[edge.edge_id] = edge
    return list(by_id.values())