Skip to content

graflo.hq.endpoint_resolve

Resolve edge endpoints declared by a secondary identity.

An edge-only source references its endpoints by an alternate key. Before the edge can be written, that key has to be mapped back to the vertex's primary identity — which is what every backend's edge write already expects.

Doing the mapping here rather than pushing the predicate into each backend's edge query keeps one semantic across all of them, and is the only approach that works for backends addressing endpoints by key (PostgreSQL foreign keys, NebulaGraph VIDs, TigerGraph PRIMARY_ID).

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

AmbiguousEndpointError

Bases: RuntimeError

A secondary identity matched several vertices under the error policy.

Source code in graflo/hq/endpoint_resolve.py 
31
32
class AmbiguousEndpointError(RuntimeError):
    """A secondary identity matched several vertices under the ``error`` policy."""

AttachStats dataclass

What happened while attaching records of one class to existing vertices.

An attached record never creates a vertex, so a record whose key finds none is counted and not written.

Source code in graflo/hq/endpoint_resolve.py 
 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
107
108
109
110
111
112
113
114
115
116
117
118
119
@dataclass
class AttachStats:
    """What happened while attaching records of one class to existing vertices.

    An attached record never creates a vertex, so a record whose key finds none
    is counted and not written.
    """

    documents: int = 0
    attached: int = 0
    """Records written onto at least one vertex."""
    unmatched: int = 0
    """Records whose key matched no vertex."""
    unresolvable: int = 0
    """Records whose key was absent or incomplete, so no lookup was possible."""
    ambiguous: int = 0
    """Records whose key matched more than one vertex."""
    written: int = 0
    """Rows upserted, one per vertex written onto: above ``attached`` when
    fanning out, below it when several records land on one vertex."""

    def has_findings(self) -> bool:
        return bool(self.unresolvable or self.unmatched or self.ambiguous)

    def summary(self) -> str:
        return (
            f"documents={self.documents} attached={self.attached} "
            f"written={self.written} unresolvable={self.unresolvable} "
            f"unmatched={self.unmatched} ambiguous={self.ambiguous}"
        )

    def merge(self, other: AttachStats) -> None:
        """Add *other*'s counts to these."""
        self.documents += other.documents
        self.attached += other.attached
        self.unmatched += other.unmatched
        self.unresolvable += other.unresolvable
        self.ambiguous += other.ambiguous
        self.written += other.written

Attributes

ambiguous = 0 class-attribute instance-attribute

Records whose key matched more than one vertex.

attached = 0 class-attribute instance-attribute

Records written onto at least one vertex.

documents = 0 class-attribute instance-attribute
unmatched = 0 class-attribute instance-attribute

Records whose key matched no vertex.

unresolvable = 0 class-attribute instance-attribute

Records whose key was absent or incomplete, so no lookup was possible.

written = 0 class-attribute instance-attribute

Rows upserted, one per vertex written onto: above attached when fanning out, below it when several records land on one vertex.

Methods:

__init__(documents=0, attached=0, unmatched=0, unresolvable=0, ambiguous=0, written=0)
has_findings()
Source code in graflo/hq/endpoint_resolve.py 
102
103
def has_findings(self) -> bool:
    return bool(self.unresolvable or self.unmatched or self.ambiguous)
merge(other)

Add other's counts to these.

Source code in graflo/hq/endpoint_resolve.py 
112
113
114
115
116
117
118
119
def merge(self, other: AttachStats) -> None:
    """Add *other*'s counts to these."""
    self.documents += other.documents
    self.attached += other.attached
    self.unmatched += other.unmatched
    self.unresolvable += other.unresolvable
    self.ambiguous += other.ambiguous
    self.written += other.written
summary()
Source code in graflo/hq/endpoint_resolve.py 
105
106
107
108
109
110
def summary(self) -> str:
    return (
        f"documents={self.documents} attached={self.attached} "
        f"written={self.written} unresolvable={self.unresolvable} "
        f"unmatched={self.unmatched} ambiguous={self.ambiguous}"
    )

EndpointResolutionStats dataclass

What happened while resolving one edge batch.

Unmatched and ambiguous endpoints are ordinary data conditions rather than failures, so they are counted and reported instead of raising.

Source code in graflo/hq/endpoint_resolve.py 
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
@dataclass
class EndpointResolutionStats:
    """What happened while resolving one edge batch.

    Unmatched and ambiguous endpoints are ordinary data conditions rather than
    failures, so they are counted and reported instead of raising.
    """

    documents: int = 0
    unresolvable: int = 0
    """Documents whose key was absent or incomplete, so no lookup was possible."""
    unmatched: int = 0
    """Documents whose key matched no vertex."""
    ambiguous: int = 0
    """Documents whose key matched more than one vertex."""
    dropped: int = 0
    """Documents that produced no edge."""
    written: int = 0
    """Edge documents produced, which exceeds ``documents`` when fanning out."""
    endpoints: list[str] = field(default_factory=list)

    def has_findings(self) -> bool:
        return bool(self.unresolvable or self.unmatched or self.ambiguous)

    def summary(self) -> str:
        return (
            f"endpoints={'+'.join(self.endpoints) or 'none'} "
            f"documents={self.documents} "
            f"written={self.written} dropped={self.dropped} "
            f"unresolvable={self.unresolvable} unmatched={self.unmatched} "
            f"ambiguous={self.ambiguous}"
        )

    def merge(self, other: EndpointResolutionStats) -> None:
        """Add *other*'s counts to these."""
        self.documents += other.documents
        self.unresolvable += other.unresolvable
        self.unmatched += other.unmatched
        self.ambiguous += other.ambiguous
        self.dropped += other.dropped
        self.written += other.written
        for endpoint in other.endpoints:
            if endpoint not in self.endpoints:
                self.endpoints.append(endpoint)

Attributes

ambiguous = 0 class-attribute instance-attribute

Documents whose key matched more than one vertex.

documents = 0 class-attribute instance-attribute
dropped = 0 class-attribute instance-attribute

Documents that produced no edge.

endpoints = field(default_factory=list) class-attribute instance-attribute
unmatched = 0 class-attribute instance-attribute

Documents whose key matched no vertex.

unresolvable = 0 class-attribute instance-attribute

Documents whose key was absent or incomplete, so no lookup was possible.

written = 0 class-attribute instance-attribute

Edge documents produced, which exceeds documents when fanning out.

Methods:

__init__(documents=0, unresolvable=0, unmatched=0, ambiguous=0, dropped=0, written=0, endpoints=list())
has_findings()
Source code in graflo/hq/endpoint_resolve.py 
56
57
def has_findings(self) -> bool:
    return bool(self.unresolvable or self.unmatched or self.ambiguous)
merge(other)

Add other's counts to these.

Source code in graflo/hq/endpoint_resolve.py 
68
69
70
71
72
73
74
75
76
77
78
def merge(self, other: EndpointResolutionStats) -> None:
    """Add *other*'s counts to these."""
    self.documents += other.documents
    self.unresolvable += other.unresolvable
    self.unmatched += other.unmatched
    self.ambiguous += other.ambiguous
    self.dropped += other.dropped
    self.written += other.written
    for endpoint in other.endpoints:
        if endpoint not in self.endpoints:
            self.endpoints.append(endpoint)
summary()
Source code in graflo/hq/endpoint_resolve.py 
59
60
61
62
63
64
65
66
def summary(self) -> str:
    return (
        f"endpoints={'+'.join(self.endpoints) or 'none'} "
        f"documents={self.documents} "
        f"written={self.written} dropped={self.dropped} "
        f"unresolvable={self.unresolvable} unmatched={self.unmatched} "
        f"ambiguous={self.ambiguous}"
    )

WriteStats dataclass

Resolution counts accumulated by one writer over a run.

Safe to add to from the worker threads concurrent writes run on.

Source code in graflo/hq/endpoint_resolve.py 
122
123
124
125
126
127
128
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
@dataclass
class WriteStats:
    """Resolution counts accumulated by one writer over a run.

    Safe to add to from the worker threads concurrent writes run on.
    """

    attached: dict[str, AttachStats] = field(default_factory=dict)
    """By vertex class."""
    endpoints: dict[EdgeId, EndpointResolutionStats] = field(default_factory=dict)
    """By edge, for edges whose endpoints were resolved by a secondary identity."""
    _lock: threading.Lock = field(
        default_factory=threading.Lock, repr=False, compare=False
    )

    def add_attached(self, vertex: str, stats: AttachStats) -> None:
        with self._lock:
            self.attached.setdefault(vertex, AttachStats()).merge(stats)

    def add_endpoints(self, edge_id: EdgeId, stats: EndpointResolutionStats) -> None:
        with self._lock:
            self.endpoints.setdefault(edge_id, EndpointResolutionStats()).merge(stats)

    def is_empty(self) -> bool:
        return not self.attached and not self.endpoints

    def summary(self) -> str:
        """One line per attached class and per resolved edge."""
        with self._lock:
            lines = [
                f"attached {vertex}: {stats.summary()}"
                for vertex, stats in self.attached.items()
            ]
            lines.extend(
                f"edge {edge_id}: {stats.summary()}"
                for edge_id, stats in self.endpoints.items()
            )
        return "\n".join(lines)

Attributes

attached = field(default_factory=dict) class-attribute instance-attribute

By vertex class.

endpoints = field(default_factory=dict) class-attribute instance-attribute

By edge, for edges whose endpoints were resolved by a secondary identity.

Methods:

__init__(attached=dict(), endpoints=dict(), _lock=threading.Lock())
add_attached(vertex, stats)
Source code in graflo/hq/endpoint_resolve.py 
137
138
139
def add_attached(self, vertex: str, stats: AttachStats) -> None:
    with self._lock:
        self.attached.setdefault(vertex, AttachStats()).merge(stats)
add_endpoints(edge_id, stats)
Source code in graflo/hq/endpoint_resolve.py 
141
142
143
def add_endpoints(self, edge_id: EdgeId, stats: EndpointResolutionStats) -> None:
    with self._lock:
        self.endpoints.setdefault(edge_id, EndpointResolutionStats()).merge(stats)
is_empty()
Source code in graflo/hq/endpoint_resolve.py 
145
146
def is_empty(self) -> bool:
    return not self.attached and not self.endpoints
summary()

One line per attached class and per resolved edge.

Source code in graflo/hq/endpoint_resolve.py 
148
149
150
151
152
153
154
155
156
157
158
159
def summary(self) -> str:
    """One line per attached class and per resolved edge."""
    with self._lock:
        lines = [
            f"attached {vertex}: {stats.summary()}"
            for vertex, stats in self.attached.items()
        ]
        lines.extend(
            f"edge {edge_id}: {stats.summary()}"
            for edge_id, stats in self.endpoints.items()
        )
    return "\n".join(lines)

Functions:

resolve_edge_endpoints(db, docs, *, source_class, target_class, source_match_fields, target_match_fields, source_identity_fields, target_identity_fields, resolve_source, resolve_target, policy, guard=None)

Rewrite endpoint projections from secondary keys to primary identities.

Only the endpoints flagged for resolution are looked up; the other side is passed through untouched, which is what makes asymmetric selection work.

Parameters:

Name Type Description Default
docs list[Any]

(source_projection, target_projection, weight) triples

 required
resolve_source bool

True when the source endpoint uses a secondary identity

 required
resolve_target bool

Same, for the target endpoint

 required
policy EndpointAmbiguityPolicy

What to do when a key matches several vertices

 required
guard FilterExpression | None

A trusted read guard every match must pass, so an endpoint is never resolved to a vertex the guard hides; refused by a backend without read guards.

 None

Returns:

Name Type Description
tuple tuple[list[Any], EndpointResolutionStats]

rewritten edge documents, and what happened while resolving.

Raises:

Type Description
AmbiguousEndpointError

on multiple matches under the error policy.
Source code in graflo/hq/endpoint_resolve.py 
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def resolve_edge_endpoints(
    db: Any,
    docs: list[Any],
    *,
    source_class: str,
    target_class: str,
    source_match_fields: Sequence[str],
    target_match_fields: Sequence[str],
    source_identity_fields: Sequence[str],
    target_identity_fields: Sequence[str],
    resolve_source: bool,
    resolve_target: bool,
    policy: EndpointAmbiguityPolicy,
    guard: FilterExpression | None = None,
) -> tuple[list[Any], EndpointResolutionStats]:
    """Rewrite endpoint projections from secondary keys to primary identities.

    Only the endpoints flagged for resolution are looked up; the other side is
    passed through untouched, which is what makes asymmetric selection work.

    Args:
        docs: ``(source_projection, target_projection, weight)`` triples
        resolve_source: True when the source endpoint uses a secondary identity
        resolve_target: Same, for the target endpoint
        policy: What to do when a key matches several vertices
        guard: A trusted read guard every match must pass, so an endpoint is
            never resolved to a vertex the guard hides; refused by a backend
            without read guards.

    Returns:
        tuple: rewritten edge documents, and what happened while resolving.

    Raises:
        AmbiguousEndpointError: on multiple matches under the ``error`` policy.
    """
    stats = EndpointResolutionStats(documents=len(docs))
    if resolve_source:
        stats.endpoints.append("source")
    if resolve_target:
        stats.endpoints.append("target")

    # Only passed when set, so a backend that predates guards is untouched.
    guarded: dict[str, Any] = {"guard": guard} if guard is not None else {}
    source_matches: dict[int, list[dict[str, Any]]] = {}
    target_matches: dict[int, list[dict[str, Any]]] = {}
    if resolve_source:
        source_matches = db.resolve_vertices(
            source_class,
            [doc[0] for doc in docs],
            tuple(source_match_fields),
            tuple(source_identity_fields),
            **guarded,
        )
    if resolve_target:
        target_matches = db.resolve_vertices(
            target_class,
            [doc[1] for doc in docs],
            tuple(target_match_fields),
            tuple(target_identity_fields),
            **guarded,
        )

    resolved: list[Any] = []
    for position, doc in enumerate(docs):
        source_doc, target_doc = doc[0], doc[1]
        rest = tuple(doc[2:])

        source_options = _candidates_for(
            position=position,
            projection=source_doc,
            matches=source_matches,
            match_fields=source_match_fields,
            identity_fields=source_identity_fields,
            resolve=resolve_source,
            stats=stats,
            side="source",
            source_class=source_class,
            policy=policy,
        )
        target_options = _candidates_for(
            position=position,
            projection=target_doc,
            matches=target_matches,
            match_fields=target_match_fields,
            identity_fields=target_identity_fields,
            resolve=resolve_target,
            stats=stats,
            side="target",
            source_class=target_class,
            policy=policy,
        )

        if not source_options or not target_options:
            stats.dropped += 1
            continue

        for resolved_source in source_options:
            for resolved_target in target_options:
                resolved.append((resolved_source, resolved_target, *rest))
                stats.written += 1

    return resolved, stats

select_matches(found, *, policy, identity_fields, describe)

The vertices a key's matches resolve to under policy.

Several matches are kept (all), reduced to the lowest primary identity (first), dropped (skip) or refused (error); a single match is always kept. Callers check for a partial key before looking it up.

Parameters:

Name Type Description Default
found list[dict[str, Any]]

Every vertex the key matched, carrying identity_fields

 required
policy EndpointAmbiguityPolicy

What to do when there are several matches

 required
identity_fields Sequence[str]

Primary identity fields to project the matches onto

 required
describe Callable[[int], str]

Given the match count, the sentence naming what matched, for the error message

 required

Returns:

Name Type Description
list list[dict[str, Any]]

The selected matches, projected onto identity_fields.

Raises:

Type Description
AmbiguousEndpointError

on several matches under the error policy.
Source code in graflo/hq/endpoint_resolve.py 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def select_matches(
    found: list[dict[str, Any]],
    *,
    policy: EndpointAmbiguityPolicy,
    identity_fields: Sequence[str],
    describe: Callable[[int], str],
) -> list[dict[str, Any]]:
    """The vertices a key's matches resolve to under *policy*.

    Several matches are kept (``all``), reduced to the lowest primary identity
    (``first``), dropped (``skip``) or refused (``error``); a single match is
    always kept. Callers check for a partial key before looking it up.

    Args:
        found: Every vertex the key matched, carrying *identity_fields*
        policy: What to do when there are several matches
        identity_fields: Primary identity fields to project the matches onto
        describe: Given the match count, the sentence naming what matched,
            for the ``error`` message

    Returns:
        list: The selected matches, projected onto *identity_fields*.

    Raises:
        AmbiguousEndpointError: on several matches under the ``error`` policy.
    """
    if len(found) > 1:
        if policy == "error":
            raise AmbiguousEndpointError(
                f"{describe(len(found))} "
                "Set endpoints_on_ambiguous to all, first or skip to tolerate this."
            )
        if policy == "skip":
            return []
        if policy == "first":
            found = _sorted_candidates(found, identity_fields)[:1]

    return [
        {field_name: doc.get(field_name) for field_name in identity_fields}
        for doc in found
    ]