graflo.db.edge_direction_support

Backend support for edge directionality (advisory — never raises).

Edge.directed is a statement about the model: when false, endpoint order carries no meaning and the two orientations denote one relationship. Backends express that to wildly different degrees, and the difference matters mostly on the read path — reaching an edge from its target endpoint is free on some backends, needs an unemitted clause on others, and is a schema-time decision that cannot be retrofitted on TigerGraph.

This module records that matrix so callers (schema application, traversal planning, capability reporting) can consult one table instead of re-deriving per-backend behaviour inline.

Unlike :mod:graflo.db.field_type_support, nothing here raises. directed=False is already expressible in shipped manifests and is silently ignored by seven of the eight targets; refusing it now would reject working schemas. Callers get diagnostics and decide.

EdgeDirectionDiagnostic dataclass

One finding about how a backend will treat a logically undirected edge.

Source code in graflo/db/edge_direction_support.py

@dataclass(frozen=True)
class EdgeDirectionDiagnostic:
    """One finding about how a backend will treat a logically undirected edge."""

    edge_id: EdgeId
    db_type: DBType
    severity: Literal["info", "warning"]
    message: str
    remedy: str

    def __str__(self) -> str:
        return f"{self.message} {self.remedy}"

ReverseTraversalCost

Bases: StrEnum

What it costs to reach an edge from its target endpoint.

Source code in graflo/db/edge_direction_support.py

class ReverseTraversalCost(StrEnum):
    """What it costs to reach an edge from its *target* endpoint."""

    FREE = "free"
    """Both endpoints are indexed; the reverse query is the same price."""

    CHEAP = "cheap"
    """Relationships are stored bidirectionally; the reverse pattern is legal and fast."""

    CLAUSE_REQUIRED = "clause_required"
    """Cheap once asked for, but only via an explicit reverse/bidirectional clause."""

    INDEX_REQUIRED = "index_required"
    """Needs a secondary index on the target column before it is affordable."""

    SCHEMA_TIME_ONLY = "schema_time_only"
    """Decided at DDL time; no query rewrite can recover it afterwards."""

    MATERIALIZATION_REQUIRED = "materialization_required"
    """Direction is the storage partition key; the reverse view must be written out."""

CHEAP = 'cheap' class-attribute instance-attribute

Relationships are stored bidirectionally; the reverse pattern is legal and fast.

CLAUSE_REQUIRED = 'clause_required' class-attribute instance-attribute

Cheap once asked for, but only via an explicit reverse/bidirectional clause.

FREE = 'free' class-attribute instance-attribute

Both endpoints are indexed; the reverse query is the same price.

INDEX_REQUIRED = 'index_required' class-attribute instance-attribute

Needs a secondary index on the target column before it is affordable.

MATERIALIZATION_REQUIRED = 'materialization_required' class-attribute instance-attribute

Direction is the storage partition key; the reverse view must be written out.

SCHEMA_TIME_ONLY = 'schema_time_only' class-attribute instance-attribute

Decided at DDL time; no query rewrite can recover it afterwards.

UnsupportedEdgeDirectionError

Bases: ValueError

Raised when a backend cannot answer a read in the requested direction.

Only TigerGraph can reach this: reverse reachability there is fixed when the edge type is created (WITH REVERSE_EDGE), so no query rewrite recovers it. Failing loudly is deliberate — silently returning outgoing edges for an ANY request would under-report the neighbourhood with no signal.

Source code in graflo/db/edge_direction_support.py

class UnsupportedEdgeDirectionError(ValueError):
    """Raised when a backend cannot answer a read in the requested direction.

    Only TigerGraph can reach this: reverse reachability there is fixed when the
    edge type is created (``WITH REVERSE_EDGE``), so no query rewrite recovers it.
    Failing loudly is deliberate — silently returning outgoing edges for an
    ``ANY`` request would under-report the neighbourhood with no signal.
    """

assert_direction_supported(db_type, direction, *, has_reverse_edge=False, edge_is_undirected=False)

Raise if db_type cannot answer a read in direction.

Parameters:

Name	Type	Description	Default
db_type	DBType	Backend being queried.	required
direction	EdgeDirection	Requested orientation.	required
has_reverse_edge	bool	Whether a paired reverse edge type is declared for the edge (EdgePhysicalSpec.reverse_edge). Only consulted on backends whose reverse reachability is decided at schema time.	False
edge_is_undirected	bool	Whether the edge type itself was created undirected. On a backend with native undirected edges that already answers both orientations, so no reverse type is needed.	False

Raises:

Type	Description
UnsupportedEdgeDirectionError	when the backend physically cannot follow the edge backwards.

Source code in graflo/db/edge_direction_support.py

def assert_direction_supported(
    db_type: DBType,
    direction: EdgeDirection,
    *,
    has_reverse_edge: bool = False,
    edge_is_undirected: bool = False,
) -> None:
    """Raise if ``db_type`` cannot answer a read in ``direction``.

    Args:
        db_type: Backend being queried.
        direction: Requested orientation.
        has_reverse_edge: Whether a paired reverse edge type is declared for the
            edge (``EdgePhysicalSpec.reverse_edge``). Only consulted on backends
            whose reverse reachability is decided at schema time.
        edge_is_undirected: Whether the edge type itself was created undirected.
            On a backend with native undirected edges that already answers both
            orientations, so no reverse type is needed.

    Raises:
        UnsupportedEdgeDirectionError: when the backend physically cannot follow
            the edge backwards.
    """
    if direction is EdgeDirection.OUT:
        return
    coerced = _coerce(db_type)
    if (
        _REVERSE_TRAVERSAL_COST.get(coerced)
        is not ReverseTraversalCost.SCHEMA_TIME_ONLY
    ):
        return
    if has_reverse_edge:
        return
    if edge_is_undirected and coerced in _UNDIRECTED_NATIVE_DBS:
        return
    raise UnsupportedEdgeDirectionError(
        f"Backend '{_label(db_type)}' cannot read edges with direction "
        f"'{direction.value}': reverse reachability is fixed when the edge type "
        "is created and no query rewrite recovers it. Declare the edge type as "
        "undirected (`directed: false`), or pair it with a reverse edge type via "
        "`db_profile.edge_specs[*].reverse_edge`."
    )

check_schema_edge_directions(db_type, schema)

Report how db_type will treat each logically undirected edge.

Returns an empty list when the schema declares no undirected edges, or when the backend represents them natively. Never raises: an unknown backend yields no diagnostics rather than blocking a schema application.

Source code in graflo/db/edge_direction_support.py

def check_schema_edge_directions(
    db_type: DBType, schema: Schema
) -> list[EdgeDirectionDiagnostic]:
    """Report how ``db_type`` will treat each logically undirected edge.

    Returns an empty list when the schema declares no undirected edges, or when
    the backend represents them natively. Never raises: an unknown backend
    yields no diagnostics rather than blocking a schema application.
    """
    coerced = _coerce(db_type)
    if coerced not in _REVERSE_TRAVERSAL_COST:
        return []
    if coerced in _UNDIRECTED_NATIVE_DBS:
        return []

    cost = _REVERSE_TRAVERSAL_COST[coerced]
    effect, remedy = _UNDIRECTED_FALLBACK[cost]
    severity: Literal["info", "warning"] = (
        "warning"
        if cost
        in (
            ReverseTraversalCost.INDEX_REQUIRED,
            ReverseTraversalCost.MATERIALIZATION_REQUIRED,
        )
        else "info"
    )
    label = _label(db_type)
    return [
        EdgeDirectionDiagnostic(
            edge_id=edge_id,
            db_type=coerced,
            severity=severity,
            message=(
                f"Edge {edge_id!r} is declared undirected, but backend '{label}' has "
                f"no undirected edge type: it {effect}."
            ),
            remedy=remedy,
        )
        for edge_id in iter_undirected_edges(schema)
    ]

default_direction_for_edge(edge)

The direction a read should follow for edge when none is requested.

This is where Edge.directed stops being an annotation and starts steering queries: an undirected edge reads as :attr:EdgeDirection.ANY, because both orientations denote the same relationship and anchoring on source alone would drop half the neighbourhood.

Source code in graflo/db/edge_direction_support.py

def default_direction_for_edge(edge: Edge) -> EdgeDirection:
    """The direction a read should follow for ``edge`` when none is requested.

    This is where ``Edge.directed`` stops being an annotation and starts
    steering queries: an undirected edge reads as :attr:`EdgeDirection.ANY`,
    because both orientations denote the same relationship and anchoring on
    ``source`` alone would drop half the neighbourhood.
    """
    return EdgeDirection.OUT if edge.directed else EdgeDirection.ANY

iter_undirected_edges(schema)

Yield the id of every edge in schema declared logically undirected.

Source code in graflo/db/edge_direction_support.py

def iter_undirected_edges(schema: Schema) -> Iterable[EdgeId]:
    """Yield the id of every edge in ``schema`` declared logically undirected."""
    for edge in schema.core_schema.edge_config.values():
        if not edge.directed:
            yield edge.edge_id

reverse_traversal_cost(db_type)

What it costs to reach an edge from its target endpoint on db_type.

Raises:

Type	Description
KeyError	if db_type is not a supported write target.

Source code in graflo/db/edge_direction_support.py

def reverse_traversal_cost(db_type: DBType) -> ReverseTraversalCost:
    """What it costs to reach an edge from its target endpoint on ``db_type``.

    Raises:
        KeyError: if ``db_type`` is not a supported write target.
    """
    coerced = _coerce(db_type)
    try:
        return _REVERSE_TRAVERSAL_COST[coerced]
    except KeyError:
        raise KeyError(
            f"No reverse-traversal cost recorded for backend '{_label(db_type)}'. "
            "Every target backend must have an entry."
        ) from None

supports_native_undirected(db_type)

Whether the backend has an undirected edge type in its schema language.

Source code in graflo/db/edge_direction_support.py

def supports_native_undirected(db_type: DBType) -> bool:
    """Whether the backend has an undirected edge *type* in its schema language."""
    return _coerce(db_type) in _UNDIRECTED_NATIVE_DBS