graflo.architecture.graph_types.edge_derivation

Edge-derivation wiring shared by contract (authoring) and pipeline (runtime).

:class:EdgeDerivation is the declarative model set on edge pipeline steps (:class:~graflo.architecture.contract.ingestion.steps.models.EdgeActorConfig): how an edge step binds to extracted locations and documents. These fields do not belong in schema edge_config / :class:~graflo.architecture.schema.edge.Edge.

:class:EdgeDerivationRegistry is the mutable runtime store for ingestion-time edge behavior keyed by :class:~graflo.architecture.graph_types.identifiers.EdgeId (typically one instance per :class:~graflo.architecture.pipeline.runtime.resource.ResourceRuntime). When :attr:EdgeDerivation.relation_from_key is true, the registry records the edge id so :class:~graflo.architecture.schema.db_aware.EdgeConfigDBAware (with overlay) can align TigerGraph DDL with runtime.

An endpoint selector names the identity an edge endpoint is matched on. It is plain -- None / "identity" for the primary, a secondary identity's name, or a field list -- or per class: {Class: plain}, for an endpoint a vertex_router fills with rows of several classes. A class the mapping does not name is matched on its primary identity. :func:selector_for reduces either form to the plain selector one concrete class uses.

Attributes

EndpointSelector = PlainSelector | dict[str, PlainSelector] module-attribute

A plain selector, or one per endpoint class ({Class: plain}).

PlainSelector = str | list[str] module-attribute

A secondary identity's name, a field list, "identity" or "secondary".

Classes

EdgeDerivation

Bases: ConfigBaseModel

How this edge step selects vertex locations and reads per-document relation from data.

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

class EdgeDerivation(ConfigBaseModel):
    """How this edge step selects vertex locations and reads per-document relation from data."""

    match_source: str | None = Field(
        default=None,
        description="Require this path segment in source vertex locations.",
    )
    match_target: str | None = Field(
        default=None,
        description="Require this path segment in target vertex locations.",
    )
    exclude_source: str | None = Field(
        default=None,
        description="Exclude source locations containing this path segment.",
    )
    exclude_target: str | None = Field(
        default=None,
        description="Exclude target locations containing this path segment.",
    )
    match: str | None = Field(
        default=None,
        description="Require this segment in both source and target locations.",
    )
    relation_field: str | None = Field(
        default=None,
        description="Document/ctx field name for per-document relationship label when schema relation is unset.",
    )
    relation_from_key: bool = Field(
        default=False,
        description="If True, derive the per-document relation label from the location key during assembly.",
    )
    relation_map: dict[str, str] | None = Field(
        default=None,
        description="Raw ``relation_field`` values -> relation names.",
    )
    relation_map_only: bool = Field(
        default=False,
        description="If True, a ``relation_field`` value absent from ``relation_map`` writes no edge.",
    )
    source_match: EndpointSelector | None = Field(
        default=None,
        description=(
            "Identity selector for the source endpoint: None/'identity' for the "
            "primary identity, a secondary identity name / field list, or a "
            "mapping of endpoint class to one of those."
        ),
    )
    target_match: EndpointSelector | None = Field(
        default=None,
        description="Identity selector for the target endpoint; see source_match.",
    )
    on_ambiguous: EndpointAmbiguityPolicy | None = Field(
        default=None,
        description=(
            "Per-step override of ingestion_model.endpoints_on_ambiguous when a "
            "secondary identity matches several vertices."
        ),
    )

    emit_inverse: bool = Field(
        default=False,
        description=(
            "If True, assembly also writes the declared inverse of every edge this "
            "step writes, when that inverse edge is declared."
        ),
    )

    def uses_secondary_identity(self) -> bool:
        """True when either endpoint is matched on something other than the primary identity."""
        return not (_is_primary(self.source_match) and _is_primary(self.target_match))

    def map_relation(self, raw: Any) -> tuple[bool, Any]:
        """Apply ``relation_map`` to a value read from ``relation_field``.

        Returns:
            ``(keep, relation)``; ``keep`` is False when ``relation_map_only``
            drops the value.
        """
        relation_map = self.relation_map or {}
        if isinstance(raw, str) and raw in relation_map:
            return True, relation_map[raw]
        if self.relation_map_only:
            return False, None
        return True, raw

    def is_empty(self) -> bool:
        if self.relation_from_key or self.emit_inverse:
            return False
        return all(
            getattr(self, name) is None
            for name in (
                "match_source",
                "match_target",
                "exclude_source",
                "exclude_target",
                "match",
                "relation_field",
                "source_match",
                "target_match",
                "on_ambiguous",
            )
        )

Attributes

emit_inverse = Field(default=False, description='If True, assembly also writes the declared inverse of every edge this step writes, when that inverse edge is declared.') class-attribute instance-attribute

exclude_source = Field(default=None, description='Exclude source locations containing this path segment.') class-attribute instance-attribute

exclude_target = Field(default=None, description='Exclude target locations containing this path segment.') class-attribute instance-attribute

match = Field(default=None, description='Require this segment in both source and target locations.') class-attribute instance-attribute

match_source = Field(default=None, description='Require this path segment in source vertex locations.') class-attribute instance-attribute

match_target = Field(default=None, description='Require this path segment in target vertex locations.') class-attribute instance-attribute

on_ambiguous = Field(default=None, description='Per-step override of ingestion_model.endpoints_on_ambiguous when a secondary identity matches several vertices.') class-attribute instance-attribute

relation_field = Field(default=None, description='Document/ctx field name for per-document relationship label when schema relation is unset.') class-attribute instance-attribute

relation_from_key = Field(default=False, description='If True, derive the per-document relation label from the location key during assembly.') class-attribute instance-attribute

relation_map = Field(default=None, description='Raw ``relation_field`` values -> relation names.') class-attribute instance-attribute

relation_map_only = Field(default=False, description='If True, a ``relation_field`` value absent from ``relation_map`` writes no edge.') class-attribute instance-attribute

source_match = Field(default=None, description="Identity selector for the source endpoint: None/'identity' for the primary identity, a secondary identity name / field list, or a mapping of endpoint class to one of those.") class-attribute instance-attribute

target_match = Field(default=None, description='Identity selector for the target endpoint; see source_match.') class-attribute instance-attribute

Methods:

is_empty()

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

def is_empty(self) -> bool:
    if self.relation_from_key or self.emit_inverse:
        return False
    return all(
        getattr(self, name) is None
        for name in (
            "match_source",
            "match_target",
            "exclude_source",
            "exclude_target",
            "match",
            "relation_field",
            "source_match",
            "target_match",
            "on_ambiguous",
        )
    )

map_relation(raw)

Apply relation_map to a value read from relation_field.

Returns:

Type	Description
bool	(keep, relation); keep is False when relation_map_only
Any	drops the value.

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

def map_relation(self, raw: Any) -> tuple[bool, Any]:
    """Apply ``relation_map`` to a value read from ``relation_field``.

    Returns:
        ``(keep, relation)``; ``keep`` is False when ``relation_map_only``
        drops the value.
    """
    relation_map = self.relation_map or {}
    if isinstance(raw, str) and raw in relation_map:
        return True, relation_map[raw]
    if self.relation_map_only:
        return False, None
    return True, raw

uses_secondary_identity()

True when either endpoint is matched on something other than the primary identity.

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

def uses_secondary_identity(self) -> bool:
    """True when either endpoint is matched on something other than the primary identity."""
    return not (_is_primary(self.source_match) and _is_primary(self.target_match))

EdgeDerivationRegistry

Mutable store for ingestion-time edge behavior keyed by :class:EdgeId.

Lives under the ingestion layer (typically one instance per :class:ResourceRuntime), not on :class:~graflo.architecture.schema.core.CoreSchema.

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

class EdgeDerivationRegistry:
    """Mutable store for ingestion-time edge behavior keyed by :class:`EdgeId`.

    Lives under the ingestion layer (typically one instance per :class:`ResourceRuntime`),
    not on :class:`~graflo.architecture.schema.core.CoreSchema`.
    """

    def __init__(self) -> None:
        self._relation_from_key: dict[EdgeId, bool] = {}
        self._vertex_weights: dict[EdgeId, list[Weight]] = {}
        self._endpoint_match: dict[EdgeId, EndpointMatch] = {}
        self._endpoint_rules: list[EndpointRule] = []
        self._attached: dict[str, PlainSelector] = {}

    def attach(self, vertex: str, selector: PlainSelector) -> None:
        """Record that this resource finds *vertex* by *selector*.

        Every edge endpoint of that class is then matched on it, unless an edge
        step states a selector for that endpoint.
        """
        self._attached[vertex] = selector

    def attached_selector(self, vertex: str) -> PlainSelector | None:
        """The selector *vertex* is found by, or ``None`` when it is not attached."""
        return self._attached.get(vertex)

    def mark_relation_from_key(self, edge_id: EdgeId) -> None:
        self._relation_from_key[edge_id] = True

    def uses_relation_from_key(self, edge_id: EdgeId) -> bool:
        return self._relation_from_key.get(edge_id, False)

    def set_endpoint_match(self, edge_id: EdgeId, match: EndpointMatch) -> None:
        """Record how *edge_id* locates its endpoints, for the write stage.

        Only non-default selections are stored, so an edge matching on primary
        identity — the overwhelming majority — leaves no entry and costs nothing.
        """
        if match.is_default():
            return
        self._endpoint_match[edge_id] = match

    def add_endpoint_rule(self, rule: EndpointRule) -> None:
        """Record selectors for the edges a data-driven edge step can write.

        A later rule covering the same edge wins, as a later
        :meth:`set_endpoint_match` does.
        """
        if rule in self._endpoint_rules:
            return
        self._endpoint_rules.append(rule)

    def endpoint_match_for(self, edge_id: EdgeId) -> EndpointMatch | None:
        """How *edge_id* locates its endpoints, or ``None`` for the primary identity.

        An edge registered by id wins; otherwise the last rule covering it
        decides, reduced to the edge's concrete classes. Per endpoint, one
        still on the primary identity whose class is attached takes the
        attached selector.
        """
        match = self._endpoint_match.get(edge_id)
        if match is None:
            for rule in reversed(self._endpoint_rules):
                if rule.applies_to(edge_id):
                    match = rule.resolve(edge_id)
                    break
        if self._attached:
            source, target, _relation = edge_id
            base = match or EndpointMatch()
            match = EndpointMatch(
                source=self._attached.get(source, base.source)
                if _is_primary(base.source)
                else base.source,
                target=self._attached.get(target, base.target)
                if _is_primary(base.target)
                else base.target,
                on_ambiguous=base.on_ambiguous,
            )
        if match is None or match.is_default():
            return None
        return match

    def has_endpoint_matches(self) -> bool:
        """Whether any edge locates its endpoints by a secondary identity.

        Such edges -- and attached vertices -- are resolved against database
        state at write time, which makes cross-batch write ordering semantic
        for the resource.
        """
        return bool(self._endpoint_match or self._endpoint_rules or self._attached)

    def merge_vertex_weights(self, edge_id: EdgeId, rules: list[Weight]) -> None:
        """Append vertex weight rules for *edge_id*, deduplicating by stable fingerprint."""
        if not rules:
            return
        bucket = self._vertex_weights.setdefault(edge_id, [])
        seen = {_weight_fingerprint(w) for w in bucket}
        for w in rules:
            fp = _weight_fingerprint(w)
            if fp in seen:
                continue
            seen.add(fp)
            bucket.append(w)

    def vertex_weights_for(self, edge_id: EdgeId) -> list[Weight]:
        return list(self._vertex_weights.get(edge_id, ()))

    def copy(self) -> EdgeDerivationRegistry:
        out = EdgeDerivationRegistry()
        out._relation_from_key = dict(self._relation_from_key)
        out._vertex_weights = {
            k: [w.model_copy(deep=True) for w in v]
            for k, v in self._vertex_weights.items()
        }
        out._endpoint_match = dict(self._endpoint_match)
        out._endpoint_rules = list(self._endpoint_rules)
        out._attached = dict(self._attached)
        return out

    def merge_from(self, other: EdgeDerivationRegistry) -> None:
        for eid, flag in other._relation_from_key.items():
            if flag:
                self.mark_relation_from_key(eid)
        for eid, weights in other._vertex_weights.items():
            self.merge_vertex_weights(eid, weights)
        for eid, match in other._endpoint_match.items():
            self.set_endpoint_match(eid, match)
        for rule in other._endpoint_rules:
            self.add_endpoint_rule(rule)
        for vertex, selector in other._attached.items():
            self.attach(vertex, selector)

Methods:

__init__()

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

def __init__(self) -> None:
    self._relation_from_key: dict[EdgeId, bool] = {}
    self._vertex_weights: dict[EdgeId, list[Weight]] = {}
    self._endpoint_match: dict[EdgeId, EndpointMatch] = {}
    self._endpoint_rules: list[EndpointRule] = []
    self._attached: dict[str, PlainSelector] = {}

add_endpoint_rule(rule)

Record selectors for the edges a data-driven edge step can write.

A later rule covering the same edge wins, as a later :meth:set_endpoint_match does.

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

def add_endpoint_rule(self, rule: EndpointRule) -> None:
    """Record selectors for the edges a data-driven edge step can write.

    A later rule covering the same edge wins, as a later
    :meth:`set_endpoint_match` does.
    """
    if rule in self._endpoint_rules:
        return
    self._endpoint_rules.append(rule)

attach(vertex, selector)

Record that this resource finds vertex by selector.

Every edge endpoint of that class is then matched on it, unless an edge step states a selector for that endpoint.

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

def attach(self, vertex: str, selector: PlainSelector) -> None:
    """Record that this resource finds *vertex* by *selector*.

    Every edge endpoint of that class is then matched on it, unless an edge
    step states a selector for that endpoint.
    """
    self._attached[vertex] = selector

attached_selector(vertex)

The selector vertex is found by, or None when it is not attached.

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

def attached_selector(self, vertex: str) -> PlainSelector | None:
    """The selector *vertex* is found by, or ``None`` when it is not attached."""
    return self._attached.get(vertex)

copy()

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

def copy(self) -> EdgeDerivationRegistry:
    out = EdgeDerivationRegistry()
    out._relation_from_key = dict(self._relation_from_key)
    out._vertex_weights = {
        k: [w.model_copy(deep=True) for w in v]
        for k, v in self._vertex_weights.items()
    }
    out._endpoint_match = dict(self._endpoint_match)
    out._endpoint_rules = list(self._endpoint_rules)
    out._attached = dict(self._attached)
    return out

endpoint_match_for(edge_id)

How edge_id locates its endpoints, or None for the primary identity.

An edge registered by id wins; otherwise the last rule covering it decides, reduced to the edge's concrete classes. Per endpoint, one still on the primary identity whose class is attached takes the attached selector.

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

def endpoint_match_for(self, edge_id: EdgeId) -> EndpointMatch | None:
    """How *edge_id* locates its endpoints, or ``None`` for the primary identity.

    An edge registered by id wins; otherwise the last rule covering it
    decides, reduced to the edge's concrete classes. Per endpoint, one
    still on the primary identity whose class is attached takes the
    attached selector.
    """
    match = self._endpoint_match.get(edge_id)
    if match is None:
        for rule in reversed(self._endpoint_rules):
            if rule.applies_to(edge_id):
                match = rule.resolve(edge_id)
                break
    if self._attached:
        source, target, _relation = edge_id
        base = match or EndpointMatch()
        match = EndpointMatch(
            source=self._attached.get(source, base.source)
            if _is_primary(base.source)
            else base.source,
            target=self._attached.get(target, base.target)
            if _is_primary(base.target)
            else base.target,
            on_ambiguous=base.on_ambiguous,
        )
    if match is None or match.is_default():
        return None
    return match

has_endpoint_matches()

Whether any edge locates its endpoints by a secondary identity.

Such edges -- and attached vertices -- are resolved against database state at write time, which makes cross-batch write ordering semantic for the resource.

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

def has_endpoint_matches(self) -> bool:
    """Whether any edge locates its endpoints by a secondary identity.

    Such edges -- and attached vertices -- are resolved against database
    state at write time, which makes cross-batch write ordering semantic
    for the resource.
    """
    return bool(self._endpoint_match or self._endpoint_rules or self._attached)

mark_relation_from_key(edge_id)

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

def mark_relation_from_key(self, edge_id: EdgeId) -> None:
    self._relation_from_key[edge_id] = True

merge_from(other)

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

def merge_from(self, other: EdgeDerivationRegistry) -> None:
    for eid, flag in other._relation_from_key.items():
        if flag:
            self.mark_relation_from_key(eid)
    for eid, weights in other._vertex_weights.items():
        self.merge_vertex_weights(eid, weights)
    for eid, match in other._endpoint_match.items():
        self.set_endpoint_match(eid, match)
    for rule in other._endpoint_rules:
        self.add_endpoint_rule(rule)
    for vertex, selector in other._attached.items():
        self.attach(vertex, selector)

merge_vertex_weights(edge_id, rules)

Append vertex weight rules for edge_id, deduplicating by stable fingerprint.

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

def merge_vertex_weights(self, edge_id: EdgeId, rules: list[Weight]) -> None:
    """Append vertex weight rules for *edge_id*, deduplicating by stable fingerprint."""
    if not rules:
        return
    bucket = self._vertex_weights.setdefault(edge_id, [])
    seen = {_weight_fingerprint(w) for w in bucket}
    for w in rules:
        fp = _weight_fingerprint(w)
        if fp in seen:
            continue
        seen.add(fp)
        bucket.append(w)

set_endpoint_match(edge_id, match)

Record how edge_id locates its endpoints, for the write stage.

Only non-default selections are stored, so an edge matching on primary identity — the overwhelming majority — leaves no entry and costs nothing.

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

def set_endpoint_match(self, edge_id: EdgeId, match: EndpointMatch) -> None:
    """Record how *edge_id* locates its endpoints, for the write stage.

    Only non-default selections are stored, so an edge matching on primary
    identity — the overwhelming majority — leaves no entry and costs nothing.
    """
    if match.is_default():
        return
    self._endpoint_match[edge_id] = match

uses_relation_from_key(edge_id)

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

def uses_relation_from_key(self, edge_id: EdgeId) -> bool:
    return self._relation_from_key.get(edge_id, False)

vertex_weights_for(edge_id)

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

def vertex_weights_for(self, edge_id: EdgeId) -> list[Weight]:
    return list(self._vertex_weights.get(edge_id, ()))

EndpointMatch dataclass

Which identity each endpoint of an edge is matched on at write time.

None selectors mean the vertex's primary identity, which is the default and leaves the write path exactly as it has always been.

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

@dataclass(frozen=True)
class EndpointMatch:
    """Which identity each endpoint of an edge is matched on at write time.

    ``None`` selectors mean the vertex's primary identity, which is the default
    and leaves the write path exactly as it has always been.
    """

    source: str | list[str] | None = None
    target: str | list[str] | None = None
    on_ambiguous: EndpointAmbiguityPolicy | None = None

    def is_default(self) -> bool:
        return (
            self.source in (None, PRIMARY_IDENTITY_SELECTOR)
            and self.target in (None, PRIMARY_IDENTITY_SELECTOR)
            and self.on_ambiguous is None
        )

Attributes

on_ambiguous = None class-attribute instance-attribute

source = None class-attribute instance-attribute

target = None class-attribute instance-attribute

Methods:

__init__(source=None, target=None, on_ambiguous=None)

is_default()

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

def is_default(self) -> bool:
    return (
        self.source in (None, PRIMARY_IDENTITY_SELECTOR)
        and self.target in (None, PRIMARY_IDENTITY_SELECTOR)
        and self.on_ambiguous is None
    )

EndpointRule dataclass

Endpoint selectors for edges an edge step names only per document.

A step whose endpoint comes from a router role, or whose relation comes from the data, has no single edge id to register at load time. The rule stands for every edge it can write: None in :attr:source, :attr:target or :attr:relation is whatever the document resolves, and a per-class selector is reduced to the concrete class when the rule is applied.

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

@dataclass(frozen=True)
class EndpointRule:
    """Endpoint selectors for edges an edge step names only per document.

    A step whose endpoint comes from a router role, or whose relation comes
    from the data, has no single edge id to register at load time. The rule
    stands for every edge it can write: ``None`` in :attr:`source`,
    :attr:`target` or :attr:`relation` is whatever the document resolves, and
    a per-class selector is reduced to the concrete class when the rule is
    applied.
    """

    source: str | None
    target: str | None
    relation: str | None
    source_match: EndpointSelector | None = None
    target_match: EndpointSelector | None = None
    on_ambiguous: EndpointAmbiguityPolicy | None = None

    def applies_to(self, edge_id: EdgeId) -> bool:
        return all(
            fixed is None or fixed == actual
            for fixed, actual in zip((self.source, self.target, self.relation), edge_id)
        )

    def resolve(self, edge_id: EdgeId) -> EndpointMatch:
        source, target, _relation = edge_id
        return EndpointMatch(
            source=selector_for(self.source_match, source),
            target=selector_for(self.target_match, target),
            on_ambiguous=self.on_ambiguous,
        )

Attributes

on_ambiguous = None class-attribute instance-attribute

relation instance-attribute

source instance-attribute

source_match = None class-attribute instance-attribute

target instance-attribute

target_match = None class-attribute instance-attribute

Methods:

__init__(source, target, relation, source_match=None, target_match=None, on_ambiguous=None)

applies_to(edge_id)

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

def applies_to(self, edge_id: EdgeId) -> bool:
    return all(
        fixed is None or fixed == actual
        for fixed, actual in zip((self.source, self.target, self.relation), edge_id)
    )

resolve(edge_id)

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

def resolve(self, edge_id: EdgeId) -> EndpointMatch:
    source, target, _relation = edge_id
    return EndpointMatch(
        source=selector_for(self.source_match, source),
        target=selector_for(self.target_match, target),
        on_ambiguous=self.on_ambiguous,
    )

Functions:

selector_for(selector, vertex)

The plain selector selector applies to endpoint class vertex.

A per-class mapping that does not name vertex leaves it on its primary identity (None).

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

def selector_for(
    selector: EndpointSelector | None, vertex: str
) -> PlainSelector | None:
    """The plain selector *selector* applies to endpoint class *vertex*.

    A per-class mapping that does not name *vertex* leaves it on its primary
    identity (``None``).
    """
    if isinstance(selector, dict):
        return selector.get(vertex)
    return selector