graflo.architecture.graph_types.transform

Transform payloads and observation merging.

TransformPayload

Bases: ConfigBaseModel

Typed transform output shared between extraction and assembly phases.

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

class TransformPayload(ConfigBaseModel):
    """Typed transform output shared between extraction and assembly phases."""

    named: dict[str, Any] = Field(default_factory=dict)
    positional: tuple[Any, ...] = Field(default_factory=tuple)
    removed_keys: frozenset[str] = Field(default_factory=frozenset)

    @classmethod
    def from_result(cls, result: Any) -> TransformPayload:
        if isinstance(result, dict):
            return cls(named=dict(result))
        if isinstance(result, tuple):
            return cls(positional=tuple(result))
        return cls(positional=(result,))

    def context_doc(self) -> dict[str, Any]:
        """Named values available for weight and relation extraction."""
        return dict(self.named)

context_doc()

Named values available for weight and relation extraction.

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

def context_doc(self) -> dict[str, Any]:
    """Named values available for weight and relation extraction."""
    return dict(self.named)

VertexRep

Bases: ConfigBaseModel

Context for graph transformation actions.

Attributes:

Name	Type	Description
vertex	dict[str, Any]	doc representing a vertex

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

class VertexRep(ConfigBaseModel):
    """Context for graph transformation actions.

    Attributes:
        vertex: doc representing a vertex
    """

    model_config = ConfigDict(kw_only=True)  # ty: ignore[invalid-key]

    vertex: dict[str, Any]
    lookup_only: bool = False
    """True when this observation exists only to locate an existing vertex.

    Such documents take part in edge rendering but are never upserted. Tagging
    the observation rather than the vertex type keeps it correct when one
    resource both writes and merely references the same vertex type.
    """

lookup_only = False class-attribute instance-attribute

True when this observation exists only to locate an existing vertex.

Such documents take part in edge rendering but are never upserted. Tagging the observation rather than the vertex type keeps it correct when one resource both writes and merely references the same vertex type.

context_dict_from_transform_buffer_item(item)

Map one transform_buffer entry to a flat context dict (named keys only).

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

def context_dict_from_transform_buffer_item(item: Any) -> dict[str, Any]:
    """Map one ``transform_buffer`` entry to a flat context dict (named keys only)."""
    if isinstance(item, TransformPayload):
        return item.context_doc()
    if isinstance(item, dict):
        return dict(item)
    return {}

merge_observation_with_transform_buffer(observation, buffer_items)

Merge a JSON observation slice with transform outputs at the same location.

observation is the current dict-shaped fragment of the nested document passed into actors (often a child object under a :class:DescendActor). buffer_items are the entries in ExtractionContext.transform_buffer for the same :class:LocationIndex.

Starts from a shallow copy of observation; each buffer entry (in pipeline order) updates the merged view, so later transforms override earlier keys and transform output overrides the raw JSON on key conflicts.

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

def merge_observation_with_transform_buffer(
    observation: dict[str, Any],
    buffer_items: list[Any],
) -> dict[str, Any]:
    """Merge a JSON observation slice with transform outputs at the same location.

    ``observation`` is the current dict-shaped fragment of the nested document
    passed into actors (often a child object under a :class:`DescendActor`).
    ``buffer_items`` are the entries in ``ExtractionContext.transform_buffer``
    for the same :class:`LocationIndex`.

    Starts from a shallow copy of ``observation``; each buffer entry (in pipeline
    order) updates the merged view, so later transforms override earlier keys
    and transform output overrides the raw JSON on key conflicts.
    """
    merged: dict[str, Any] = dict(observation)
    for item in buffer_items:
        merged.update(context_dict_from_transform_buffer_item(item))
        if isinstance(item, TransformPayload) and item.removed_keys:
            for k in item.removed_keys:
                merged.pop(k, None)
    return merged

merge_row_doc_with_transform_buffer(doc, buffer_items)

Backward-compatible alias for :func:merge_observation_with_transform_buffer.

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

def merge_row_doc_with_transform_buffer(
    doc: dict[str, Any],
    buffer_items: list[Any],
) -> dict[str, Any]:
    """Backward-compatible alias for :func:`merge_observation_with_transform_buffer`."""
    return merge_observation_with_transform_buffer(doc, buffer_items)