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

AmbiguousEndpointError

Bases: RuntimeError

A secondary identity matched several vertices under the error policy.

Source code in graflo/hq/endpoint_resolve.py

class AmbiguousEndpointError(RuntimeError):
    """A secondary identity matched several vertices under the ``error`` policy."""

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

@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.
    """

    rows: int = 0
    unresolvable: int = 0
    """Rows whose key was absent or incomplete, so no lookup was possible."""
    unmatched: int = 0
    """Rows whose key matched no vertex."""
    ambiguous: int = 0
    """Rows whose key matched more than one vertex."""
    dropped: int = 0
    """Rows that produced no edge."""
    written: int = 0
    """Edge documents produced, which exceeds rows 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'} rows={self.rows} "
            f"written={self.written} dropped={self.dropped} "
            f"unresolvable={self.unresolvable} unmatched={self.unmatched} "
            f"ambiguous={self.ambiguous}"
        )

ambiguous = 0 class-attribute instance-attribute

Rows whose key matched more than one vertex.

dropped = 0 class-attribute instance-attribute

Rows that produced no edge.

unmatched = 0 class-attribute instance-attribute

Rows whose key matched no vertex.

unresolvable = 0 class-attribute instance-attribute

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

written = 0 class-attribute instance-attribute

Edge documents produced, which exceeds rows when fanning out.

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)

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

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

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

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

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

    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),
        )
    if resolve_target:
        target_matches = db.resolve_vertices(
            target_class,
            [doc[1] for doc in docs],
            tuple(target_match_fields),
            tuple(target_identity_fields),
        )

    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