Skip to content

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.

EdgeDerivation

Bases: ConfigBaseModel

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

Source code in graflo/architecture/graph_types/edge_derivation.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class EdgeDerivation(ConfigBaseModel):
    """How this edge step selects vertex locations and reads per-row 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-row relationship label when schema relation is unset.",
    )
    relation_from_key: bool = Field(
        default=False,
        description="If True, derive the per-row relation label from the location key during assembly.",
    )
    source_match: str | list[str] | None = Field(
        default=None,
        description=(
            "Identity selector for the source endpoint: None/'identity' for the "
            "primary identity, or a secondary identity name / field list."
        ),
    )
    target_match: str | list[str] | 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."
        ),
    )

    def uses_secondary_identity(self) -> bool:
        """True when either endpoint is matched on something other than the primary identity."""
        return any(
            selector not in (None, PRIMARY_IDENTITY_SELECTOR)
            for selector in (self.source_match, self.target_match)
        )

    def is_empty(self) -> bool:
        if self.relation_from_key:
            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",
            )
        )

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 
83
84
85
86
87
88
def uses_secondary_identity(self) -> bool:
    """True when either endpoint is matched on something other than the primary identity."""
    return any(
        selector not in (None, PRIMARY_IDENTITY_SELECTOR)
        for selector in (self.source_match, 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 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
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] = {}

    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 endpoint_match_for(self, edge_id: EdgeId) -> EndpointMatch | None:
        return self._endpoint_match.get(edge_id)

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

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

    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)
        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)

has_endpoint_matches()

Whether any edge locates its endpoints by a secondary identity.

Such edges 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 
160
161
162
163
164
165
166
def has_endpoint_matches(self) -> bool:
    """Whether any edge locates its endpoints by a secondary identity.

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

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 
168
169
170
171
172
173
174
175
176
177
178
179
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 
147
148
149
150
151
152
153
154
155
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

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 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@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
        )