graflo.architecture.contract.ingestion.resource

Declarative resource configuration (YAML/manifest contract).

Attributes

Resource = ResourceConfig module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

EdgeInferSpec

Bases: ConfigBaseModel

Selector for controlling inferred edge emission.

Source code in graflo/architecture/contract/ingestion/resource.py

class EdgeInferSpec(ConfigBaseModel):
    """Selector for controlling inferred edge emission."""

    source: str = PydanticField(..., description="Edge source vertex name.")
    target: str = PydanticField(..., description="Edge target vertex name.")
    relation: str | None = PydanticField(
        default=None,
        description=(
            "Optional relation discriminator. If omitted, selector applies to all relations "
            "for (source, target)."
        ),
    )

    @property
    def edge_id(self) -> EdgeId:
        return self.source, self.target, self.relation

    def matches(self, edge_id: EdgeId) -> bool:
        source, target, relation = edge_id
        return (
            self.source == source
            and self.target == target
            and (self.relation is None or self.relation == relation)
        )

Attributes

edge_id property

relation = PydanticField(default=None, description='Optional relation discriminator. If omitted, selector applies to all relations for (source, target).') class-attribute instance-attribute

source = PydanticField(..., description='Edge source vertex name.') class-attribute instance-attribute

target = PydanticField(..., description='Edge target vertex name.') class-attribute instance-attribute

Methods:

matches(edge_id)

Source code in graflo/architecture/contract/ingestion/resource.py

def matches(self, edge_id: EdgeId) -> bool:
    source, target, relation = edge_id
    return (
        self.source == source
        and self.target == target
        and (self.relation is None or self.relation == relation)
    )

ResourceConfig

Bases: ConfigBaseModel

Declarative resource definition (serializable contract).

Source code in graflo/architecture/contract/ingestion/resource.py

class ResourceConfig(ConfigBaseModel):
    """Declarative resource definition (serializable contract)."""

    model_config = {"extra": "forbid"}

    name: str = PydanticField(
        ...,
        description="Name of the resource (e.g. table or file identifier).",
    )
    pipeline: list[dict[str, Any]] = PydanticField(
        ...,
        description="Pipeline of actor steps to apply in sequence (vertex, edge, transform, descend). "
        'Each step is a dict, e.g. {"vertex": "user"} or {"edge": {"from": "a", "to": "b"}}.',
        validation_alias=AliasChoices("pipeline", "apply"),
    )
    encoding: EncodingType = PydanticField(
        default=EncodingType.UTF_8,
        description="Character encoding for input/output (e.g. utf-8, ISO-8859-1).",
    )
    merge_collections: list[str] = PydanticField(
        default_factory=list,
        description=(
            "Collection names whose documents fuse when written to the graph -- "
            "several observations becoming one node, not two type declarations "
            "becoming one. Named `merge_` because it is an authored contract key; "
            "the vocabulary calls this sense `fuse`."
        ),
    )
    extra_weights: list[ResourceExtraWeightEntry] = PydanticField(
        default_factory=list,
        description="Additional edge attribute / vertex-weight enrichment for this resource.",
    )
    types: dict[str, str] = PydanticField(
        default_factory=dict,
        description='Field name to Python type expression for casting (e.g. {"amount": "float"}).',
    )
    infer_edges: bool = PydanticField(
        default=True,
        description=(
            "If True, infer edges from current vertex population. "
            "If False, emit only edges explicitly declared as edge actors in the pipeline."
        ),
    )
    infer_edge_only: list[EdgeInferSpec] = PydanticField(
        default_factory=list,
        description=(
            "Optional allow-list for inferred edges. Applies only to inferred (greedy) edges, "
            "not explicit edge actors."
        ),
    )
    infer_edge_except: list[EdgeInferSpec] = PydanticField(
        default_factory=list,
        description=(
            "Optional deny-list for inferred edges. Applies only to inferred (greedy) edges, "
            "not explicit edge actors."
        ),
    )
    drop_trivial_input_fields: bool = PydanticField(
        default=False,
        description=(
            "If True, remove top-level input keys whose value is None or the empty string before "
            "the actor pipeline runs."
        ),
    )
    fail_fast: bool = PydanticField(
        default=False,
        description=(
            "If True, a transform step fails when required input keys are missing in the "
            "current document (rename: all source keys must be present; call: all input keys). "
            "If False (default), rename applies only to keys present in the document and "
            "functional transforms skip the step when inputs are missing."
        ),
    )
    tolerate_transform_errors: bool = PydanticField(
        default=True,
        description=(
            "If True, a failing transform step sets its declared output fields to None, "
            "records the error, and continues the pipeline."
        ),
    )

    @model_validator(mode="after")
    def _validate_policy(self) -> ResourceConfig:
        if self.infer_edge_only and self.infer_edge_except:
            raise ValueError(
                "Resource infer_edge_only and infer_edge_except are mutually exclusive."
            )
        return self

    def collect_vertex_names(self) -> set[str]:
        """Vertex types referenced by this resource (pipeline and related config)."""
        names = collect_vertex_names_from_pipeline(self.pipeline)
        names.update(self.merge_collections)
        for spec in self.infer_edge_only:
            names.add(spec.source)
            names.add(spec.target)
        for spec in self.infer_edge_except:
            names.add(spec.source)
            names.add(spec.target)
        for entry in self.extra_weights:
            names.add(entry.edge.source)
            names.add(entry.edge.target)
            for weight in entry.vertex_weights:
                if weight.name is not None:
                    names.add(weight.name)
        return names

    def canonical_field_payload(self, field_name: str) -> Any | None:
        """Canonical rendering of *field_name*, when it differs from its dump.

        Consulted by content hashing and by the manifest differ. ``pipeline``
        is stored as authored dicts, and a step has several equivalent
        spellings; this renders each step in one spelling (see
        :func:`~graflo.architecture.contract.ingestion.steps.parse.canonical_actor_step`).
        ``None`` means the field's ordinary dump is already canonical.
        """
        if field_name != "pipeline":
            return None
        from graflo.architecture.contract.ingestion.steps.parse import (
            canonical_actor_step,
        )

        return [canonical_actor_step(step) for step in self.pipeline]

    def pipeline_actor_count(self) -> int:
        """Count actors in the pipeline without binding schema context."""
        from graflo.architecture.pipeline.runtime.actor import ActorWrapper

        return ActorWrapper(*self.pipeline).count()

Attributes

drop_trivial_input_fields = PydanticField(default=False, description='If True, remove top-level input keys whose value is None or the empty string before the actor pipeline runs.') class-attribute instance-attribute

encoding = PydanticField(default=EncodingType.UTF_8, description='Character encoding for input/output (e.g. utf-8, ISO-8859-1).') class-attribute instance-attribute

extra_weights = PydanticField(default_factory=list, description='Additional edge attribute / vertex-weight enrichment for this resource.') class-attribute instance-attribute

fail_fast = PydanticField(default=False, description='If True, a transform step fails when required input keys are missing in the current document (rename: all source keys must be present; call: all input keys). If False (default), rename applies only to keys present in the document and functional transforms skip the step when inputs are missing.') class-attribute instance-attribute

infer_edge_except = PydanticField(default_factory=list, description='Optional deny-list for inferred edges. Applies only to inferred (greedy) edges, not explicit edge actors.') class-attribute instance-attribute

infer_edge_only = PydanticField(default_factory=list, description='Optional allow-list for inferred edges. Applies only to inferred (greedy) edges, not explicit edge actors.') class-attribute instance-attribute

infer_edges = PydanticField(default=True, description='If True, infer edges from current vertex population. If False, emit only edges explicitly declared as edge actors in the pipeline.') class-attribute instance-attribute

merge_collections = PydanticField(default_factory=list, description='Collection names whose documents fuse when written to the graph -- several observations becoming one node, not two type declarations becoming one. Named `merge_` because it is an authored contract key; the vocabulary calls this sense `fuse`.') class-attribute instance-attribute

model_config = {'extra': 'forbid'} class-attribute instance-attribute

name = PydanticField(..., description='Name of the resource (e.g. table or file identifier).') class-attribute instance-attribute

pipeline = PydanticField(..., description='Pipeline of actor steps to apply in sequence (vertex, edge, transform, descend). Each step is a dict, e.g. {"vertex": "user"} or {"edge": {"from": "a", "to": "b"}}.', validation_alias=AliasChoices('pipeline', 'apply')) class-attribute instance-attribute

tolerate_transform_errors = PydanticField(default=True, description='If True, a failing transform step sets its declared output fields to None, records the error, and continues the pipeline.') class-attribute instance-attribute

types = PydanticField(default_factory=dict, description='Field name to Python type expression for casting (e.g. {"amount": "float"}).') class-attribute instance-attribute

Methods:

canonical_field_payload(field_name)

Canonical rendering of field_name, when it differs from its dump.

Consulted by content hashing and by the manifest differ. pipeline is stored as authored dicts, and a step has several equivalent spellings; this renders each step in one spelling (see :func:~graflo.architecture.contract.ingestion.steps.parse.canonical_actor_step). None means the field's ordinary dump is already canonical.

Source code in graflo/architecture/contract/ingestion/resource.py

def canonical_field_payload(self, field_name: str) -> Any | None:
    """Canonical rendering of *field_name*, when it differs from its dump.

    Consulted by content hashing and by the manifest differ. ``pipeline``
    is stored as authored dicts, and a step has several equivalent
    spellings; this renders each step in one spelling (see
    :func:`~graflo.architecture.contract.ingestion.steps.parse.canonical_actor_step`).
    ``None`` means the field's ordinary dump is already canonical.
    """
    if field_name != "pipeline":
        return None
    from graflo.architecture.contract.ingestion.steps.parse import (
        canonical_actor_step,
    )

    return [canonical_actor_step(step) for step in self.pipeline]

collect_vertex_names()

Vertex types referenced by this resource (pipeline and related config).

Source code in graflo/architecture/contract/ingestion/resource.py

def collect_vertex_names(self) -> set[str]:
    """Vertex types referenced by this resource (pipeline and related config)."""
    names = collect_vertex_names_from_pipeline(self.pipeline)
    names.update(self.merge_collections)
    for spec in self.infer_edge_only:
        names.add(spec.source)
        names.add(spec.target)
    for spec in self.infer_edge_except:
        names.add(spec.source)
        names.add(spec.target)
    for entry in self.extra_weights:
        names.add(entry.edge.source)
        names.add(entry.edge.target)
        for weight in entry.vertex_weights:
            if weight.name is not None:
                names.add(weight.name)
    return names

pipeline_actor_count()

Count actors in the pipeline without binding schema context.

Source code in graflo/architecture/contract/ingestion/resource.py

def pipeline_actor_count(self) -> int:
    """Count actors in the pipeline without binding schema context."""
    from graflo.architecture.pipeline.runtime.actor import ActorWrapper

    return ActorWrapper(*self.pipeline).count()

ResourceExtraWeightEntry

Bases: ConfigBaseModel

Schema edge plus optional vertex-derived weight rules for DB enrichment.

Source code in graflo/architecture/contract/ingestion/resource.py

class ResourceExtraWeightEntry(ConfigBaseModel):
    """Schema edge plus optional vertex-derived weight rules for DB enrichment."""

    edge: Edge
    vertex_weights: list[Weight] = PydanticField(default_factory=list)

    @model_validator(mode="before")
    @classmethod
    def _from_yaml(cls, data: Any) -> Any:
        if data is None:
            return data
        if isinstance(data, Edge):
            return {"edge": data, "vertex_weights": []}
        if not isinstance(data, dict):
            raise TypeError(
                f"extra_weights item must be dict or Edge, got {type(data)}"
            )
        d = dict(data)
        vw_raw = d.pop("vertex_weights", None) or []
        if not isinstance(vw_raw, list):
            vw_raw = [vw_raw]
        v_w = [Weight.model_validate(x) for x in vw_raw]
        if "edge" in d and isinstance(d["edge"], dict):
            edge = Edge.model_validate(dict(d.pop("edge")))
            if d:
                raise ValueError(
                    f"extra_weights entry has unexpected keys with 'edge': {sorted(d)}"
                )
            return {"edge": edge, "vertex_weights": v_w}
        edge = Edge.model_validate(d)
        return {"edge": edge, "vertex_weights": v_w}

Attributes

edge instance-attribute

vertex_weights = PydanticField(default_factory=list) class-attribute instance-attribute

Functions:

collect_vertex_names_from_pipeline(steps)

Collect vertex names referenced by pipeline steps (including nested descend).

Source code in graflo/architecture/contract/ingestion/resource.py

def collect_vertex_names_from_pipeline(steps: list[Any]) -> set[str]:
    """Collect vertex names referenced by pipeline steps (including nested descend)."""
    names: set[str] = set()
    for step in steps:
        if not isinstance(step, dict):
            continue
        normalized = normalize_actor_step(dict(step))
        step_type = normalized.get("type")
        if step_type == "vertex" and isinstance(normalized.get("vertex"), str):
            names.add(normalized["vertex"])
        elif step_type == "vertex_router":
            type_map = normalized.get("type_map")
            if isinstance(type_map, dict):
                for value in type_map.values():
                    if isinstance(value, str):
                        names.add(value)
            vertex_from_map = normalized.get("vertex_from_map")
            if isinstance(vertex_from_map, dict):
                for key in vertex_from_map:
                    if isinstance(key, str):
                        names.add(key)
        elif step_type == "edge":
            source = normalized.get("source") or normalized.get("from")
            target = normalized.get("target") or normalized.get("to")
            if isinstance(source, str):
                names.add(source)
            if isinstance(target, str):
                names.add(target)
            vertex_weights = normalized.get("vertex_weights")
            if isinstance(vertex_weights, list):
                for weight in vertex_weights:
                    if isinstance(weight, dict) and isinstance(weight.get("name"), str):
                        names.add(weight["name"])
        elif step_type == "descend":
            sub_pipeline = normalized.get("pipeline")
            if isinstance(sub_pipeline, list):
                names |= collect_vertex_names_from_pipeline(sub_pipeline)
    return names

find_vertex_producing_levels(steps, vertex, *, known_vertices=None)

Index paths of every pipeline level with a step producing vertex.

A path indexes one level's steps per element, descending through descend steps: [] is the root level, [2] the level inside the root's third step, [2, 0] one further down. Paths are returned outermost-first.

This is how a level-targeted op finds where to act. The level matters because an actor reads its transform buffer at its own LocationIndex with no ancestor fallback, so a derivation appended at the root is invisible to a vertex produced under a descend.

Two tiers. Levels with an explicit producer — a vertex step or a router whose table names the class — decide when any exist. Only when none does, and known_vertices declares the class, every level holding a vertex_router counts: the router routes the raw discriminator value as the class name, which is the whole mechanism of a router without a type_map. An explicit table outranks pass-through so that adding one dynamic router elsewhere never turns a resolved level ambiguous.

Source code in graflo/architecture/contract/ingestion/resource.py

def find_vertex_producing_levels(
    steps: list[Any], vertex: str, *, known_vertices: Collection[str] | None = None
) -> list[list[int]]:
    """Index paths of every pipeline level with a step producing *vertex*.

    A path indexes one level's steps per element, descending through ``descend``
    steps: ``[]`` is the root level, ``[2]`` the level inside the root's third
    step, ``[2, 0]`` one further down. Paths are returned outermost-first.

    This is how a level-targeted op finds where to act. The level matters
    because an actor reads its transform buffer at its own ``LocationIndex``
    with no ancestor fallback, so a derivation appended at the root is invisible
    to a vertex produced under a ``descend``.

    Two tiers. Levels with an *explicit* producer — a ``vertex`` step or a
    router whose table names the class — decide when any exist. Only when none
    does, and *known_vertices* declares the class, every level holding a
    ``vertex_router`` counts: the router routes the raw discriminator value
    as the class name, which is the whole mechanism of a router without a
    ``type_map``. An explicit table outranks pass-through so that adding one
    dynamic router elsewhere never turns a resolved level ambiguous.
    """
    explicit = _walk_levels(steps, lambda level: _level_produces(level, vertex))
    if explicit or known_vertices is None or vertex not in known_vertices:
        return explicit
    return _walk_levels(steps, _level_has_router)

pipeline_has_vertex_router(steps)

Whether any level of steps holds a vertex_router.

A router routes an unmapped discriminator value as the class name, so a pipeline holding one can produce any class the schema declares — not only the names its steps state. Anything scoping a schema to a resource by the names its pipeline mentions must widen to every class when this is true, or the router silently drops each record whose class it did not name.

Source code in graflo/architecture/contract/ingestion/resource.py

def pipeline_has_vertex_router(steps: list[Any]) -> bool:
    """Whether any level of *steps* holds a ``vertex_router``.

    A router routes an unmapped discriminator value as the class name, so a
    pipeline holding one can produce any class the schema declares — not only
    the names its steps state. Anything scoping a schema to a resource by the
    names its pipeline mentions must widen to every class when this is true,
    or the router silently drops each record whose class it did not name.
    """
    return bool(_walk_levels(steps, _level_has_router))

resolve_pipeline_level(steps, path)

Return the live step list path addresses inside steps.

Mutating the returned list mutates steps. Getting that guarantee requires rewriting each walked descend step into its normalized form and storing it back: the shorthand spellings ({descend: {apply: [...]}}, a bare {key, apply}) keep their sub-steps under a different key, so returning whatever normalize_actor_step built would hand back a list nothing holds — and an append into it would vanish without a word. Steps off the path, and the root level itself, are left exactly as authored.

Raises when the path does not resolve. Every index on the way must address a descend step; those are the only steps that own a nested level.

Source code in graflo/architecture/contract/ingestion/resource.py

def resolve_pipeline_level(steps: list[Any], path: list[int]) -> list[Any]:
    """Return the live step list *path* addresses inside *steps*.

    Mutating the returned list mutates *steps*. Getting that guarantee requires
    rewriting each walked ``descend`` step into its normalized form and storing
    it back: the shorthand spellings (``{descend: {apply: [...]}}``, a bare
    ``{key, apply}``) keep their sub-steps under a different key, so returning
    whatever ``normalize_actor_step`` built would hand back a list nothing
    holds — and an append into it would vanish without a word. Steps off the
    path, and the root level itself, are left exactly as authored.

    Raises when the path does not resolve. Every index on the way must address
    a ``descend`` step; those are the only steps that own a nested level.
    """
    level = steps
    for depth, index in enumerate(path):
        walked = path[:depth] or "root"
        if not 0 <= index < len(level):
            raise ValueError(
                f"pipeline path {path} does not resolve: index {index} is out of "
                f"range at level {walked}"
            )
        step = level[index]
        if not isinstance(step, dict):
            raise ValueError(
                f"pipeline path {path} does not resolve: step {index} at level "
                f"{walked} is not an actor step"
            )
        normalized = normalize_actor_step(dict(step))
        if normalized.get("type") != "descend":
            raise ValueError(
                f"pipeline path {path} does not resolve: step {index} at level "
                f"{walked} is a {normalized.get('type')!r} step, not a descend — "
                "only a descend owns a nested level"
            )
        sub_pipeline = normalized.get("pipeline")
        normalized["pipeline"] = sub_pipeline if isinstance(sub_pipeline, list) else []
        level[index] = normalized
        level = normalized["pipeline"]
    return level

step_produces_vertices(step, *, known_vertices=None)

Vertex names a single (non-recursive) actor step produces.

Production, not reference: an edge step names endpoints it looks up, so it is not counted. A vertex_router produces every type its type_map can select and every type its vertex_from_map projects — its explicit targets. A router also routes an unmapped discriminator value as-is, as the class name, so with known_vertices (the schema's declared classes) it produces every one of them by pass-through as well: that is how a router without a type_map works at all, and the static picture must not say it produces nothing.

Source code in graflo/architecture/contract/ingestion/resource.py

def step_produces_vertices(
    step: dict[str, Any], *, known_vertices: Collection[str] | None = None
) -> set[str]:
    """Vertex names a single (non-recursive) actor step *produces*.

    Production, not reference: an ``edge`` step names endpoints it looks up, so
    it is not counted. A ``vertex_router`` produces every type its ``type_map``
    can select and every type its ``vertex_from_map`` projects — its *explicit*
    targets. A router also routes an unmapped discriminator value as-is, as the
    class name, so with *known_vertices* (the schema's declared classes) it
    produces every one of them by pass-through as well: that is how a router
    without a ``type_map`` works at all, and the static picture must not say
    it produces nothing.
    """
    normalized = normalize_actor_step(dict(step))
    step_type = normalized.get("type")
    if step_type == "vertex" and isinstance(normalized.get("vertex"), str):
        return {normalized["vertex"]}
    if step_type == "vertex_router":
        names = _router_explicit_targets(normalized)
        if known_vertices is not None:
            names |= set(known_vertices)
        return names
    return set()