graflo.db.traversal

Backend-neutral multi-hop traversal, composed from single-hop primitives.

Every backend that can answer fetch_edges gets correct multi-hop semantics from this module, in one shape: a :class:~graflo.architecture.graph_types.container.GraphContainer. Backends with a native multi-hop query override :meth:Connection.graph_neighbors for a single round trip, and the conformance suite asserts the override agrees with this default rather than merely "returning something".

Direction is decided per edge, not per request: an edge declared directed=False is followed both ways whatever the caller asked for, and a backend that physically cannot follow the requested direction fails loudly before any query runs, rather than returning a partial neighbourhood.

Attributes

DEFAULT_EDGE_LIMIT = 1000 module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

bfs_neighbors(conn, *, anchor_type, anchor_key, hops=1, direction=EdgeDirection.OUT, edge_types=None, filters=None, limit=None, schema=None)

Breadth-first neighbourhood around one anchor, as a GraphContainer.

Parameters:

Name	Type	Description	Default
conn	Connection	Live connection. Only fetch_edges and fetch_docs are used — never execute, which must stay off any agent-reachable path.	required
anchor_type	str	Logical vertex type of the anchor.	required
anchor_key	str | dict[str, Any]	Anchor identity value, or a field mapping to resolve.	required
hops	int	Maximum hop distance, at least 1.	1
direction	EdgeDirection	Orientations followed from each frontier vertex.	OUT
edge_types	Sequence[str] | None	Restrict to these logical relation names. A declared inverse that stores nothing is a valid name: it reads the forward edge from its target, and the result reports it under the name asked for, endpoints in that reading's order.	None
filters	Any | None	Optional edge filter, rendered per backend dialect.	None
limit	int | None	Maximum accumulated edges. Defaults to DEFAULT_EDGE_LIMIT.	None
schema	Schema | None	Required — logical names must be resolved to storage names, or the "universal layer" leaks backend naming to the caller.	None

Returns:

Name	Type	Description
GraphContainer	GraphContainer	reached vertices and edges, deduplicated.

Raises:

Type	Description
ValueError	if hops < 1, schema is missing, or anchor_type is not declared in the schema.
UnsupportedEdgeDirectionError	if a backend cannot follow an edge in the requested direction.

Source code in graflo/db/traversal.py

def bfs_neighbors(
    conn: Connection,
    *,
    anchor_type: str,
    anchor_key: str | dict[str, Any],
    hops: int = 1,
    direction: EdgeDirection = EdgeDirection.OUT,
    edge_types: Sequence[str] | None = None,
    filters: Any | None = None,
    limit: int | None = None,
    schema: Schema | None = None,
) -> GraphContainer:
    """Breadth-first neighbourhood around one anchor, as a ``GraphContainer``.

    Args:
        conn: Live connection. Only ``fetch_edges`` and ``fetch_docs`` are used —
            never ``execute``, which must stay off any agent-reachable path.
        anchor_type: Logical vertex type of the anchor.
        anchor_key: Anchor identity value, or a field mapping to resolve.
        hops: Maximum hop distance, at least 1.
        direction: Orientations followed from each frontier vertex.
        edge_types: Restrict to these logical relation names. A declared
            inverse that stores nothing is a valid name:
            it reads the forward edge from its target, and the result reports
            it under the name asked for, endpoints in that reading's order.
        filters: Optional edge filter, rendered per backend dialect.
        limit: Maximum accumulated edges. Defaults to ``DEFAULT_EDGE_LIMIT``.
        schema: Required — logical names must be resolved to storage names, or
            the "universal layer" leaks backend naming to the caller.

    Returns:
        GraphContainer: reached vertices and edges, deduplicated.

    Raises:
        ValueError: if *hops* < 1, *schema* is missing, or *anchor_type* is not
            declared in the schema.
        UnsupportedEdgeDirectionError: if a backend cannot follow an edge in the
            requested direction.
    """
    if hops < 1:
        raise ValueError(f"hops must be >= 1, got {hops}")
    if schema is None:
        raise ValueError(
            "graph_neighbors requires a schema: logical vertex and relation names "
            "cannot be resolved to storage names without one"
        )
    if anchor_type not in schema.core_schema.vertex_config.vertex_set:
        raise ValueError(
            f"Unknown vertex type {anchor_type!r}; declared: "
            f"{sorted(schema.core_schema.vertex_config.vertex_set)}"
        )

    max_edges = DEFAULT_EDGE_LIMIT if limit is None else limit
    db_aware = schema.resolve_db_aware(conn.flavor)
    edge_config = schema.core_schema.edge_config
    selected = select_edges(schema, edge_types)

    container = GraphContainer()
    anchor_id = _resolve_anchor_id(conn, schema, db_aware, anchor_type, anchor_key)
    if anchor_id is None:
        return container

    visited: set[tuple[str, str]] = {(anchor_type, anchor_id)}
    frontier: list[tuple[str, str]] = [(anchor_type, anchor_id)]
    seen_edges: set[tuple[EdgeId, str]] = set()
    edge_count = 0

    for _hop in range(hops):
        if not frontier or edge_count >= max_edges:
            break
        next_frontier: list[tuple[str, str]] = []
        for current_type, current_id in frontier:
            for edge, read_as in _incident_edges(selected, current_type):
                if edge_count >= max_edges:
                    break
                # Read through its inverse name, the stored edge is followed
                # from the other end.
                effective = _edge_direction_for(
                    edge,
                    direction if read_as is None else reversed_direction(direction),
                )
                anchor_side = _anchor_side(edge, current_type, effective)
                if anchor_side is None:
                    continue
                # The paired type the database maintains for this relation, on a
                # backend where reverse reachability is a schema-time decision.
                native_inverse_type = db_aware.db_profile.native_inverse_of(
                    edge.relation, edge_config
                )
                # Assert before querying: a backend that cannot follow this
                # orientation must fail rather than silently return the half it
                # can answer.
                assert_direction_supported(
                    conn.flavor,
                    anchor_side,
                    has_native_inverse=native_inverse_type is not None,
                    edge_is_undirected=not edge.directed,
                )
                rows = _fetch_edge_rows(
                    conn,
                    db_aware=db_aware,
                    edge=edge,
                    anchor_type=current_type,
                    anchor_id=current_id,
                    direction=anchor_side,
                    filters=filters,
                    remaining=max_edges - edge_count,
                    native_inverse_type=native_inverse_type,
                )
                if not rows:
                    continue
                far_type = _far_endpoint(edge.edge_id, current_type)
                # Reported under the name asked for: whether the inverse is
                # stored, maintained by the database or only declared, one
                # question gets one shape of answer.
                edge_id = (
                    edge.edge_id
                    if read_as is None
                    else (edge.target, edge.source, read_as)
                )
                bucket = container.edges.setdefault(edge_id, [])
                far_ids: list[str] = []
                for row in rows:
                    properties, source_key, target_key = normalize_edge_row(row)
                    marker = (edge_id, _row_marker(properties, source_key, target_key))
                    if marker in seen_edges:
                        continue
                    seen_edges.add(marker)
                    # Normalize the endpoints into the row so a consumer reads
                    # one shape whatever backend answered.
                    near, far_end = (
                        (source_key, target_key)
                        if read_as is None
                        else (target_key, source_key)
                    )
                    bucket.append({**properties, "source": near, "target": far_end})
                    edge_count += 1
                    far = target_key if source_key == current_id else source_key
                    if far is not None and far != current_id or far is not None:
                        far_ids.append(far)

                for doc in _hydrate_far_endpoints(
                    conn, db_aware, schema, far_type, far_ids
                ):
                    identity = _vertex_identity_value(conn, schema, far_type, doc)
                    if identity is None or (far_type, identity) in visited:
                        continue
                    visited.add((far_type, identity))
                    container.vertices.setdefault(far_type, []).append(doc)
                    next_frontier.append((far_type, identity))
        frontier = next_frontier

    if edge_count >= max_edges:
        logger.debug(
            "graph_neighbors hit the edge limit (%s); result is truncated", max_edges
        )
    container.pick_unique()
    return container

check_anchor_fields(schema, db_aware, vertex_type, fields)

Refuse an anchor key that names a property vertex_type does not declare.

A native traversal writes the anchor's field name into its query text, and the name arrives with the request, so an undeclared one is not merely a miss: it is an injection point. Checking against the schema closes it for every backend at once.

Raises:

Type	Description
ValueError	naming the undeclared field and the declared ones.

Source code in graflo/db/traversal.py

def check_anchor_fields(
    schema: Schema, db_aware: Any, vertex_type: str, fields: Iterable[str]
) -> None:
    """Refuse an anchor key that names a property *vertex_type* does not declare.

    A native traversal writes the anchor's field name into its query text, and
    the name arrives with the request, so an undeclared one is not merely a
    miss: it is an injection point. Checking against the schema closes it for
    every backend at once.

    Raises:
        ValueError: naming the undeclared field and the declared ones.
    """
    vertex = schema.core_schema.vertex_config[vertex_type]
    declared = set(vertex.property_names) | set(
        db_aware.vertex_config.identity_fields(vertex_type)
    )
    for field in fields:
        if field not in declared:
            raise ValueError(
                f"Cannot match {vertex_type!r} on {field!r}: not a declared "
                f"property; declared: {sorted(declared)}"
            )

edge_query_name(db_aware, edge, flavor)

The identifier a backend's read path uses for edge.

Backends name edge types in incompatible ways, and no single accessor covers them: edge_storage_name is Arango-only by construction (it builds an edge collection name and returns None elsewhere), the Cypher family and Nebula key on the relation type, and PostgreSQL keys on a derived table name. Resolving that here keeps the choice in one place instead of in every caller.

Source code in graflo/db/traversal.py

def edge_query_name(db_aware: Any, edge: Edge, flavor: Any) -> str | None:
    """The identifier a backend's read path uses for *edge*.

    Backends name edge types in incompatible ways, and no single accessor covers
    them: ``edge_storage_name`` is Arango-only by construction (it builds an edge
    *collection* name and returns ``None`` elsewhere), the Cypher family and
    Nebula key on the relation type, and PostgreSQL keys on a derived table name.
    Resolving that here keeps the choice in one place instead of in every caller.
    """
    from graflo.onto import DBType

    if flavor == DBType.POSTGRES:
        from graflo.db.postgres.target_write import edge_table_name

        return edge_table_name(edge.source, edge.target, edge.relation)

    storage = db_aware.edge_config.runtime(edge).storage_name()
    if storage is not None:
        return storage
    return db_aware.edge_config.relation_dbname(edge) or edge.relation

normalize_edge_row(row)

Reduce a backend's edge row to (properties, source key, target key).

fetch_edges predates this wave and returns whatever shape each driver finds natural: Arango yields an edge document with _from/_to, the Cypher family yields the driver's (start props, type, end props) triple, PostgreSQL yields source_id/target_id columns. Normalizing here is what lets every backend answer in one GraphContainer without changing a read path other code already depends on.

Source code in graflo/db/traversal.py

def normalize_edge_row(row: Any) -> tuple[dict[str, Any], str | None, str | None]:
    """Reduce a backend's edge row to (properties, source key, target key).

    ``fetch_edges`` predates this wave and returns whatever shape each driver
    finds natural: Arango yields an edge document with ``_from``/``_to``, the
    Cypher family yields the driver's ``(start props, type, end props)`` triple,
    PostgreSQL yields ``source_id``/``target_id`` columns. Normalizing here is
    what lets every backend answer in one ``GraphContainer`` without changing a
    read path other code already depends on.
    """
    if isinstance(row, (tuple, list)):
        # Cypher drivers render a relationship as (start, type, end).
        start = row[0] if len(row) > 0 and isinstance(row[0], dict) else {}
        end = row[2] if len(row) > 2 and isinstance(row[2], dict) else {}
        properties = row[1] if len(row) > 1 and isinstance(row[1], dict) else {}
        return (
            dict(properties),
            _first_present(start, ("id", "_key", "_id")),
            _first_present(end, ("id", "_key", "_id")),
        )
    if isinstance(row, dict):
        source = next(
            (_strip_collection(row[k]) for k in _SOURCE_KEYS if row.get(k) is not None),
            None,
        )
        target = next(
            (_strip_collection(row[k]) for k in _TARGET_KEYS if row.get(k) is not None),
            None,
        )
        if source is None and target is None:
            source = _strip_collection(row.get("_from_key"))
            target = _strip_collection(row.get("_to_key"))
        if source is None and target is None and row:
            # Neither endpoint resolved, so the caller will drop this row from
            # the neighbourhood. Silent loss reads as "no such neighbour", which
            # is indistinguishable from a correct empty result — say so instead.
            shape = tuple(sorted(str(k) for k in row))
            if shape not in _reported_unresolved:
                _reported_unresolved.add(shape)
                logger.warning(
                    "normalize_edge_row: no endpoint keys in edge row with columns "
                    "%s; rows of this shape are dropped from traversal. Expected one "
                    "of %s for the source and %s for the target.",
                    list(shape),
                    list(_SOURCE_KEYS),
                    list(_TARGET_KEYS),
                )
        return dict(row), source, target
    return {}, None, None

select_edges(schema, edge_types)

Declared edges a walk may follow, each with the inverse name it is read through.

With no allow-list every declared edge is followed as stored. A relation name is resolved through the declared inverses (:func:~graflo.architecture.schema.inverse_realization.resolve_relation): a name that labels declared edges selects them as stored, and a name that is only the declared inverse of one -- nothing stored under it -- selects those edges read backwards. The second element is that inverse name, or None for an edge read as stored.

Source code in graflo/db/traversal.py

def select_edges(
    schema: Schema, edge_types: Sequence[str] | None
) -> list[tuple[Edge, str | None]]:
    """Declared edges a walk may follow, each with the inverse name it is read through.

    With no allow-list every declared edge is followed as stored. A relation
    name is resolved through the declared inverses
    (:func:`~graflo.architecture.schema.inverse_realization.resolve_relation`):
    a name that labels declared edges selects them as stored, and a name that is
    only the declared inverse of one -- nothing stored under it -- selects those
    edges read backwards. The second element is that inverse name, or ``None``
    for an edge read as stored.
    """
    if edge_types is None:
        return [(edge, None) for edge in schema.core_schema.edge_config.edges]
    selected: list[tuple[Edge, str | None]] = []
    for name in dict.fromkeys(edge_types):
        for resolved in resolve_relation(schema, name):
            selected.append((resolved.edge, name if resolved.reversed else None))
    return selected