graflo.architecture.pipeline.runtime.actor.vertex

Vertex actor for processing vertex data.

Attributes

Classes

VertexActor

Bases: VertexProducingActor

Actor for processing vertex data.

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

class VertexActor(VertexProducingActor):
    """Actor for processing vertex data."""

    def __init__(self, config: VertexActorConfig):
        self.name = config.vertex
        self.from_doc: dict[str, str] | None = config.from_doc
        self.keep_fields: tuple[str, ...] | None = (
            tuple(config.keep_fields) if config.keep_fields else None
        )
        self.extraction_scope: Literal["full", "mapped_only"] = config.extraction_scope
        self.role: str | None = config.role
        self.lookup_only: bool = config.lookup_only
        self.find: str | None = config.find
        self.identity_optional: bool = config.identity_optional
        self.vertex_config: VertexConfig
        self.allowed_vertex_names: set[VertexName] | None = None
        self._key_fields: list[str] | None = None

    @classmethod
    def from_config(cls, config: VertexActorConfig) -> VertexActor:
        return cls(config)

    def fetch_important_items(self) -> dict[str, Any]:
        return self._fetch_items_from_dict(
            (
                "name",
                "from_doc",
                "keep_fields",
                "extraction_scope",
                "role",
                "lookup_only",
                "find",
                "identity_optional",
            )
        )

    def finish_init(self, init_ctx: ActorInitContext) -> None:
        self.vertex_config = init_ctx.vertex_config
        self.allowed_vertex_names = init_ctx.allowed_vertex_names
        if init_ctx.strict_references and self.from_doc:
            refuse_undeclared_mapping(self.vertex_config, self.name, self.from_doc)
        self._key_fields = None
        if self.vertex_config.has_vertex(self.name):
            self._key_fields = (
                self.vertex_config.identity_fields(self.name)
                if self.find is None
                else find_fields(
                    self.vertex_config,
                    self.name,
                    self.find,
                    step=f"vertex step {self.name!r}",
                )
            )
            if (
                self.identity_optional
                and self.find is None
                and (
                    self.vertex_config.is_hash_identity(self.name)
                    or self.vertex_config.is_blank(self.name)
                    or self.vertex_config.is_assigned(self.name)
                )
            ):
                raise ValueError(
                    f"vertex step {self.name!r}: identity_optional needs a key the "
                    "record carries; this vertex's identity is minted"
                )

    def _resolve_key_fields(self) -> list[str]:
        """The fields this step's documents are keyed by.

        A `find` document is keyed by the fields it is found by: they are what
        expands a multi-valued source field, what positional transform outputs
        land on, and what documents merge on. Otherwise the primary identity.
        """
        if self._key_fields is not None:
            return self._key_fields
        if self.find is None:
            return self.vertex_config.identity_fields(self.name)
        return self.vertex_config.match_fields(self.name, self.find)

    def _filter_and_aggregate_vertex_docs(
        self, docs: list[dict[str, Any]], doc: dict[str, Any]
    ) -> list[dict[str, Any]]:
        filters = self.vertex_config.filters(self.name)
        return [
            _doc for _doc in docs if all(cfilter.matches(_doc) for cfilter in filters)
        ]

    def _extract_vertex_doc_from_transformed_item(
        self,
        item: Any,
        vertex_keys: tuple[str, ...],
        index_keys: tuple[str, ...],
    ) -> dict[str, Any]:
        if isinstance(item, TransformPayload):
            doc: dict[str, Any] = {}
            consumed_named: set[str] = set()
            for k, v in item.named.items():
                if k in vertex_keys and v is not None:
                    doc[k] = v
                    consumed_named.add(k)
            for j, value in enumerate(item.positional):
                if j >= len(index_keys):
                    break
                doc[index_keys[j]] = value
            for key in consumed_named:
                item.named.pop(key, None)
            if item.positional:
                item.positional = ()
            return doc

        if isinstance(item, dict):
            doc = {}
            value_keys = sorted(
                (
                    k
                    for k in item
                    if k.startswith(ActorConstants.DRESSING_TRANSFORMED_VALUE_KEY)
                ),
                key=lambda x: int(x.rsplit("#", 1)[-1]),
            )
            for j, vkey in enumerate(value_keys):
                if j >= len(index_keys):
                    break
                doc[index_keys[j]] = item.pop(vkey)
            for vkey in vertex_keys:
                if vkey not in doc and vkey in item and item[vkey] is not None:
                    doc[vkey] = item.pop(vkey)
            return doc

        return {}

    def _process_transformed_items(
        self,
        ctx: ExtractionContext,
        lindex: LocationIndex,
        doc: dict[str, Any],
        vertex_keys: tuple[str, ...],
        index_keys: tuple[str, ...],
    ) -> list[dict[str, Any]]:
        payloads = ctx.transform_buffer[lindex]
        extracted_docs = [
            self._extract_vertex_doc_from_transformed_item(
                item, vertex_keys, index_keys
            )
            for item in payloads
        ]
        ctx.transform_buffer[lindex] = [
            item
            for item in payloads
            if not (
                isinstance(item, TransformPayload)
                and not item.named
                and not item.positional
            )
            and not (isinstance(item, dict) and not item)
        ]
        return self._filter_and_aggregate_vertex_docs(extracted_docs, doc)

    def __call__(
        self, ctx: ExtractionContext, lindex: LocationIndex, *nargs: Any, **kwargs: Any
    ) -> ExtractionContext:
        doc: dict[str, Any] = kwargs.get("doc", {})
        buffer_items: list[Any] = list(ctx.transform_buffer.get(lindex, []))
        effective_doc = merge_observation_with_transform_buffer(doc, buffer_items)
        ctx.obs_buffer[lindex] = dict(effective_doc)

        # Early-exit for disallowed vertex types.
        # This must happen before any ctx.acc_vertex[...] access.
        if (
            self.allowed_vertex_names is not None
            and self.name not in self.allowed_vertex_names
        ):
            return ctx
        if self.allowed_vertex_names is None and not self.vertex_config.has_vertex(
            self.name
        ):
            return ctx

        vertex_keys_list = self.vertex_config.property_names(self.name)
        vertex_keys: tuple[str, ...] = tuple(vertex_keys_list)

        # When a role is set the vertex is stored at a named sub-slot so that
        # multiple vertices of the same type in one observation (e.g. buyer/seller)
        # occupy distinct accumulator locations. Transforms are always read from
        # the bare observation lindex; only storage moves to the role slot.
        effective_lindex = lindex.extend((self.role, 0)) if self.role else lindex

        agg = []
        key_fields = self._resolve_key_fields()
        if self.from_doc:
            source_keys = set(self.from_doc.values())
            consumed_from_buffer = False
            for item in ctx.transform_buffer[lindex]:
                if isinstance(item, TransformPayload) and source_keys.issubset(
                    item.named
                ):
                    projected = {
                        v_f: item.named[d_f] for v_f, d_f in self.from_doc.items()
                    }
                    if any(v is not None for v in projected.values()):
                        agg.extend(explode_identity_lists(projected, key_fields))
                    for k in source_keys:
                        item.named.pop(k, None)
                    consumed_from_buffer = True
            ctx.transform_buffer[lindex] = [
                item
                for item in ctx.transform_buffer[lindex]
                if not (
                    isinstance(item, TransformPayload)
                    and not item.named
                    and not item.positional
                )
                and not (isinstance(item, dict) and not item)
            ]
            if not consumed_from_buffer:
                # A field an earlier vertex step took from the transform output is
                # still this step's to read: `from` names it explicitly.
                produced = ctx.transformed.get(lindex, {})
                projected = {
                    v_f: _read_source_field(effective_doc, produced, d_f)
                    for v_f, d_f in self.from_doc.items()
                }
                if any(v is not None for v in projected.values()):
                    agg.extend(explode_identity_lists(projected, key_fields))
            buffer_vertex_keys = tuple(k for k in vertex_keys if k not in self.from_doc)
        else:
            buffer_vertex_keys = vertex_keys

        agg.extend(
            self._process_transformed_items(
                ctx, lindex, effective_doc, buffer_vertex_keys, tuple(key_fields)
            )
        )

        if self.extraction_scope == "full":
            remaining_keys = set(vertex_keys) - set().union(*[d.keys() for d in agg])
            # When keep_fields is set, restrict passthrough to only those declared fields.
            if self.keep_fields is not None:
                remaining_keys = remaining_keys & set(self.keep_fields)
            passthrough_doc = {
                k: effective_doc.get(k) for k in remaining_keys if k in effective_doc
            }
            if passthrough_doc:
                agg.append(passthrough_doc)

        if self.vertex_config.is_hash_identity(self.name):
            # A digest vertex's key is synthetic: a value the record carries
            # under that name (a source `id` column, a mapping, a positional
            # transform output) would key the document instead of the digest.
            # A `find` document gets no digest; it carries no primary key either.
            identity_fields = self.vertex_config.identity_fields(self.name)
            for vertex_doc in agg:
                for field in identity_fields:
                    vertex_doc.pop(field, None)

        merged = fuse_doc_basis(agg, index_keys=tuple(key_fields))
        if self.identity_optional and key_fields:
            # Same emptiness rule as the caster's identity filter: None or "".
            merged = [
                m for m in merged if any(m.get(f) not in (None, "") for f in key_fields)
            ]

        for m in merged:
            vertex_rep = VertexRep(
                vertex=m, lookup_only=self.lookup_only, find=self.find
            )
            ctx.acc_vertex[self.name][effective_lindex].append(vertex_rep)
            ctx.record_vertex_observation(
                vertex_name=self.name,
                location=effective_lindex,
                vertex=vertex_rep.vertex,
                ctx={},
            )
        return ctx

    def references_vertices(self) -> set[VertexName]:
        return {self.name}

Attributes

allowed_vertex_names = None instance-attribute

extraction_scope = config.extraction_scope instance-attribute

find = config.find instance-attribute

from_doc = config.from_doc instance-attribute

identity_optional = config.identity_optional instance-attribute

keep_fields = tuple(config.keep_fields) if config.keep_fields else None instance-attribute

lookup_only = config.lookup_only instance-attribute

name = config.vertex instance-attribute

role = config.role instance-attribute

vertex_config instance-attribute

Methods:

__call__(ctx, lindex, *nargs, **kwargs)

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

def __call__(
    self, ctx: ExtractionContext, lindex: LocationIndex, *nargs: Any, **kwargs: Any
) -> ExtractionContext:
    doc: dict[str, Any] = kwargs.get("doc", {})
    buffer_items: list[Any] = list(ctx.transform_buffer.get(lindex, []))
    effective_doc = merge_observation_with_transform_buffer(doc, buffer_items)
    ctx.obs_buffer[lindex] = dict(effective_doc)

    # Early-exit for disallowed vertex types.
    # This must happen before any ctx.acc_vertex[...] access.
    if (
        self.allowed_vertex_names is not None
        and self.name not in self.allowed_vertex_names
    ):
        return ctx
    if self.allowed_vertex_names is None and not self.vertex_config.has_vertex(
        self.name
    ):
        return ctx

    vertex_keys_list = self.vertex_config.property_names(self.name)
    vertex_keys: tuple[str, ...] = tuple(vertex_keys_list)

    # When a role is set the vertex is stored at a named sub-slot so that
    # multiple vertices of the same type in one observation (e.g. buyer/seller)
    # occupy distinct accumulator locations. Transforms are always read from
    # the bare observation lindex; only storage moves to the role slot.
    effective_lindex = lindex.extend((self.role, 0)) if self.role else lindex

    agg = []
    key_fields = self._resolve_key_fields()
    if self.from_doc:
        source_keys = set(self.from_doc.values())
        consumed_from_buffer = False
        for item in ctx.transform_buffer[lindex]:
            if isinstance(item, TransformPayload) and source_keys.issubset(
                item.named
            ):
                projected = {
                    v_f: item.named[d_f] for v_f, d_f in self.from_doc.items()
                }
                if any(v is not None for v in projected.values()):
                    agg.extend(explode_identity_lists(projected, key_fields))
                for k in source_keys:
                    item.named.pop(k, None)
                consumed_from_buffer = True
        ctx.transform_buffer[lindex] = [
            item
            for item in ctx.transform_buffer[lindex]
            if not (
                isinstance(item, TransformPayload)
                and not item.named
                and not item.positional
            )
            and not (isinstance(item, dict) and not item)
        ]
        if not consumed_from_buffer:
            # A field an earlier vertex step took from the transform output is
            # still this step's to read: `from` names it explicitly.
            produced = ctx.transformed.get(lindex, {})
            projected = {
                v_f: _read_source_field(effective_doc, produced, d_f)
                for v_f, d_f in self.from_doc.items()
            }
            if any(v is not None for v in projected.values()):
                agg.extend(explode_identity_lists(projected, key_fields))
        buffer_vertex_keys = tuple(k for k in vertex_keys if k not in self.from_doc)
    else:
        buffer_vertex_keys = vertex_keys

    agg.extend(
        self._process_transformed_items(
            ctx, lindex, effective_doc, buffer_vertex_keys, tuple(key_fields)
        )
    )

    if self.extraction_scope == "full":
        remaining_keys = set(vertex_keys) - set().union(*[d.keys() for d in agg])
        # When keep_fields is set, restrict passthrough to only those declared fields.
        if self.keep_fields is not None:
            remaining_keys = remaining_keys & set(self.keep_fields)
        passthrough_doc = {
            k: effective_doc.get(k) for k in remaining_keys if k in effective_doc
        }
        if passthrough_doc:
            agg.append(passthrough_doc)

    if self.vertex_config.is_hash_identity(self.name):
        # A digest vertex's key is synthetic: a value the record carries
        # under that name (a source `id` column, a mapping, a positional
        # transform output) would key the document instead of the digest.
        # A `find` document gets no digest; it carries no primary key either.
        identity_fields = self.vertex_config.identity_fields(self.name)
        for vertex_doc in agg:
            for field in identity_fields:
                vertex_doc.pop(field, None)

    merged = fuse_doc_basis(agg, index_keys=tuple(key_fields))
    if self.identity_optional and key_fields:
        # Same emptiness rule as the caster's identity filter: None or "".
        merged = [
            m for m in merged if any(m.get(f) not in (None, "") for f in key_fields)
        ]

    for m in merged:
        vertex_rep = VertexRep(
            vertex=m, lookup_only=self.lookup_only, find=self.find
        )
        ctx.acc_vertex[self.name][effective_lindex].append(vertex_rep)
        ctx.record_vertex_observation(
            vertex_name=self.name,
            location=effective_lindex,
            vertex=vertex_rep.vertex,
            ctx={},
        )
    return ctx

__init__(config)

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

def __init__(self, config: VertexActorConfig):
    self.name = config.vertex
    self.from_doc: dict[str, str] | None = config.from_doc
    self.keep_fields: tuple[str, ...] | None = (
        tuple(config.keep_fields) if config.keep_fields else None
    )
    self.extraction_scope: Literal["full", "mapped_only"] = config.extraction_scope
    self.role: str | None = config.role
    self.lookup_only: bool = config.lookup_only
    self.find: str | None = config.find
    self.identity_optional: bool = config.identity_optional
    self.vertex_config: VertexConfig
    self.allowed_vertex_names: set[VertexName] | None = None
    self._key_fields: list[str] | None = None

fetch_important_items()

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

def fetch_important_items(self) -> dict[str, Any]:
    return self._fetch_items_from_dict(
        (
            "name",
            "from_doc",
            "keep_fields",
            "extraction_scope",
            "role",
            "lookup_only",
            "find",
            "identity_optional",
        )
    )

finish_init(init_ctx)

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

def finish_init(self, init_ctx: ActorInitContext) -> None:
    self.vertex_config = init_ctx.vertex_config
    self.allowed_vertex_names = init_ctx.allowed_vertex_names
    if init_ctx.strict_references and self.from_doc:
        refuse_undeclared_mapping(self.vertex_config, self.name, self.from_doc)
    self._key_fields = None
    if self.vertex_config.has_vertex(self.name):
        self._key_fields = (
            self.vertex_config.identity_fields(self.name)
            if self.find is None
            else find_fields(
                self.vertex_config,
                self.name,
                self.find,
                step=f"vertex step {self.name!r}",
            )
        )
        if (
            self.identity_optional
            and self.find is None
            and (
                self.vertex_config.is_hash_identity(self.name)
                or self.vertex_config.is_blank(self.name)
                or self.vertex_config.is_assigned(self.name)
            )
        ):
            raise ValueError(
                f"vertex step {self.name!r}: identity_optional needs a key the "
                "record carries; this vertex's identity is minted"
            )

from_config(config) classmethod

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

@classmethod
def from_config(cls, config: VertexActorConfig) -> VertexActor:
    return cls(config)

references_vertices()

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

def references_vertices(self) -> set[VertexName]:
    return {self.name}

Functions:

explode_identity_lists(doc, identity_fields)

One document per element when an identity field holds a list.

A multi-valued source field names several vertices, not one vertex keyed on a list. None elements are dropped, so an empty list names no vertex.

Raises:

Type	Description
ValueError	If more than one identity field holds a list; how their elements pair is not stated.

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

def explode_identity_lists(
    doc: dict[str, Any], identity_fields: Sequence[str]
) -> list[dict[str, Any]]:
    """One document per element when an identity field holds a list.

    A multi-valued source field names several vertices, not one vertex keyed on
    a list. ``None`` elements are dropped, so an empty list names no vertex.

    Raises:
        ValueError: If more than one identity field holds a list; how their
            elements pair is not stated.
    """
    listed = [name for name in identity_fields if isinstance(doc.get(name), list)]
    if not listed:
        return [doc]
    if len(listed) > 1:
        raise ValueError(
            f"Identity fields {listed} each hold a list; a vertex step can expand "
            "one multi-valued identity field, not several."
        )
    field = listed[0]
    return [{**doc, field: value} for value in doc[field] if value is not None]

find_fields(vertex_config, vertex, selector, *, step)

The fields of the secondary identity selector that step finds vertex by.

Raises:

Type	Description
ValueError	selector names the primary identity or no secondary identity vertex declares.

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

def find_fields(
    vertex_config: VertexConfig, vertex: str, selector: str, *, step: str
) -> list[str]:
    """The fields of the secondary identity *selector* that *step* finds *vertex* by.

    Raises:
        ValueError: *selector* names the primary identity or no secondary
            identity *vertex* declares.
    """
    if selector == PRIMARY_IDENTITY_SELECTOR:
        raise ValueError(
            f"{step}: find must name a secondary identity of {vertex!r}, "
            f"got {selector!r}"
        )
    try:
        return vertex_config.match_fields(vertex, selector)
    except ValueError as exc:
        raise ValueError(f"{step}: find {selector!r}: {exc}") from exc

refuse_undeclared_mapping(vertex_config, vertex, from_doc)

Refuse a from map that writes a property vertex does not declare.

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

def refuse_undeclared_mapping(
    vertex_config: VertexConfig, vertex: str, from_doc: Mapping[str, str]
) -> None:
    """Refuse a ``from`` map that writes a property *vertex* does not declare."""
    unknown = undeclared_mapping_targets(vertex_config, vertex, from_doc)
    if unknown:
        raise ValueError(
            f"vertex step {vertex!r} maps {unknown} through `from`, which "
            f"{vertex!r} does not declare as properties; declare them, or map "
            "onto declared properties"
        )

undeclared_mapping_targets(vertex_config, vertex, from_doc)

The from targets vertex does not declare as properties.

Source code in graflo/architecture/pipeline/runtime/actor/vertex.py

def undeclared_mapping_targets(
    vertex_config: VertexConfig, vertex: str, from_doc: Mapping[str, str]
) -> list[str]:
    """The ``from`` targets *vertex* does not declare as properties."""
    if not vertex_config.has_vertex(vertex):
        return []
    declared = set(vertex_config.property_names(vertex))
    return sorted(target for target in from_doc if target not in declared)