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

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

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,
        # Undirected wins: it is the weaker assertion, and treating a merged
        # undirected edge as directed would let AddInverseEdgesOp synthesize an
        # inverse that duplicates it.
        directed=a.directed and b.directed,
    )

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

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.

Identity mode is carried through the merge: blank / assigned propagate when any source declares them, and hash_identity_properties / secondary_identities are unioned. The mutual exclusions enforced by :meth:Vertex.set_identity are checked here so the failure names the merge rather than surfacing from pydantic.

Source code in graflo/architecture/evolution/merge_core.py

def merge_vertex_models(vertices: list[Vertex], into_name: str) -> Vertex:
    """Union-merge vertex definitions into a single :class:`Vertex`.

    Identity mode is carried through the merge: ``blank`` / ``assigned`` propagate when
    any source declares them, and ``hash_identity_properties`` / ``secondary_identities``
    are unioned. The mutual exclusions enforced by :meth:`Vertex.set_identity` are
    checked here so the failure names the merge rather than surfacing from pydantic.
    """
    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)

    blank_out = any(v.blank for v in vertices)
    assigned_out = any(v.assigned for v in vertices)
    if blank_out and assigned_out:
        raise ValueError(
            f"Cannot merge into vertex '{into_name}': sources mix blank and assigned "
            "identity modes, which are mutually exclusive"
        )

    hash_out: list[str] = []
    seen_hash: set[str] = set()
    for v in vertices:
        for name in v.hash_identity_properties:
            if name not in seen_hash:
                hash_out.append(name)
                seen_hash.add(name)
    if assigned_out and hash_out:
        raise ValueError(
            f"Cannot merge into vertex '{into_name}': an assigned source cannot be "
            f"merged with hash-identity sources (hash properties: {hash_out})"
        )

    funnel_out = _merge_identity_funnels(vertices, into_name)
    if funnel_out is not None:
        if hash_out:
            raise ValueError(
                f"Cannot merge into vertex '{into_name}': sources mix an identity "
                f"funnel with flat hash properties {hash_out}. Express the flat key "
                "as a funnel branch first, then merge."
            )
        if assigned_out or blank_out:
            raise ValueError(
                f"Cannot merge into vertex '{into_name}': an identity funnel cannot "
                "be merged with assigned or blank sources"
            )

    secondary_out = _merge_secondary_identities(vertices, into_name, identity_out)
    if blank_out and secondary_out:
        raise ValueError(
            f"Cannot merge into vertex '{into_name}': a blank source cannot be merged "
            "with sources declaring secondary_identities — a blank vertex has no "
            "source-visible key to match on"
        )

    return Vertex(
        name=into_name,
        properties=list(props.values()),
        identity=identity_out,
        filters=filters_out,
        description=desc_out,
        blank=blank_out,
        assigned=assigned_out,
        hash_identity_properties=hash_out,
        identity_funnel=funnel_out,
        secondary_identities=secondary_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

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

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