graflo.architecture.graph_types.merge

Document fusion: several observations of one entity becoming one document.

This is the fuse plane -- two records becoming one node -- and not either schema-level sense of "merge". Nothing here folds two declarations of a type; it folds the documents a pipeline emitted for the same entity, keyed on the fields they will be matched on, preserving order and handling both dict and :class:VertexRep objects.

Key Functions

• fuse_doc_basis: fold documents sharing an index key, preserving order
• union_members: the members a merge: union field holds after an upsert

Classes

Functions:

fuse_doc_basis(docs, index_keys, union_fields=())

fuse_doc_basis(docs: list[dict], index_keys: tuple[str, ...], union_fields: Collection[str] = ()) -> list[dict]

fuse_doc_basis(docs: list[VertexRep], index_keys: tuple[str, ...], union_fields: Collection[str] = ()) -> list[VertexRep]

Merge documents based on common index keys, preserving order.

This function merges documents that share common index key-value combinations, preserving the order of documents based on the first occurrence of each index key combination. Documents without index keys are merged into the closest preceding document with index keys. If no documents have index keys, all documents are merged into a single document.

For VertexRep objects, the merge is performed on the vertex attribute.

VertexRep objects fuse only with reps carrying the same observation tags (lookup_only, find): each tag set is fused on its own, and the groups follow one another in the order of their first rep. Fusing across tags would put writable properties under a reference, or the reverse.

Parameters:

Name	Type	Description	Default
docs	list[dict] | list[VertexRep]	Homogeneous list of documents (all dict or all VertexRep) to merge	required
index_keys	tuple[str, ...]	Tuple of key names to use for merging	required
union_fields	Collection[str]	List fields fused as :func:union_members of every document's value, the first included; every other field is later-wins.	()

Returns:

Type	Description
list[dict] | list[VertexRep]	Merged documents in order of first occurrence (same type as input)

Source code in graflo/architecture/graph_types/merge.py

def fuse_doc_basis(
    docs: list[dict] | list[VertexRep],
    index_keys: tuple[str, ...],
    union_fields: Collection[str] = (),
) -> list[dict] | list[VertexRep]:
    """Merge documents based on common index keys, preserving order.

    This function merges documents that share common index key-value combinations,
    preserving the order of documents based on the first occurrence of each index
    key combination. Documents without index keys are merged into the closest
    preceding document with index keys. If no documents have index keys, all
    documents are merged into a single document.

    For VertexRep objects, the merge is performed on the `vertex` attribute.

    VertexRep objects fuse only with reps carrying the same observation tags
    (``lookup_only``, ``find``): each tag set is fused on its own, and the
    groups follow one another in the order of their first rep. Fusing across
    tags would put writable properties under a reference, or the reverse.

    Args:
        docs: Homogeneous list of documents (all dict or all VertexRep) to merge
        index_keys: Tuple of key names to use for merging
        union_fields: List fields fused as :func:`union_members` of every
            document's value, the first included; every other field is
            later-wins.

    Returns:
        Merged documents in order of first occurrence (same type as input)
    """
    if not docs:
        return docs
    if isinstance(docs[0], VertexRep):
        groups: dict[tuple[bool, str | None], list[VertexRep]] = {}
        for rep in docs:
            if not isinstance(rep, VertexRep):
                raise TypeError(f"expected VertexRep, got {type(rep).__name__}")
            groups.setdefault((rep.lookup_only, rep.find), []).append(rep)
        fused: list[VertexRep] = []
        for group in groups.values():
            fused.extend(
                cast(list[VertexRep], _fuse_group(group, index_keys, union_fields))
            )
        return fused
    return _fuse_group(docs, index_keys, union_fields)

union_members(stored, written)

What a merge: union field holds once written is upserted onto stored.

Stored members first, then new ones, each once, in first-seen order. Null members are dropped, a null value adds nothing and a scalar is one member: the union every backend that writes one applies.

Source code in graflo/architecture/graph_types/merge.py

def union_members(stored: Any, written: Any) -> list[Any]:
    """What a ``merge: union`` field holds once *written* is upserted onto *stored*.

    Stored members first, then new ones, each once, in first-seen order. Null
    members are dropped, a null value adds nothing and a scalar is one member:
    the union every backend that writes one applies.
    """
    out: list[Any] = []
    for value in (stored, written):
        if value is None:
            continue
        for member in value if isinstance(value, list) else [value]:
            if member is not None and member not in out:
                out.append(member)
    return out