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.

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

    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 in _incident_edges(schema, current_type, edge_types=edge_types):
                if edge_count >= max_edges:
                    break
                effective = _edge_direction_for(edge, direction)
                # 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,
                    effective,
                    edge_is_undirected=not edge.directed,
                )
                anchor_side = _anchor_side(edge, current_type, effective)
                if anchor_side is None:
                    continue
                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,
                )
                if not rows:
                    continue
                edge_id = edge.edge_id
                far_type = _far_endpoint(edge_id, current_type)
                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.
                    bucket.append(
                        {**properties, "source": source_key, "target": target_key}
                    )
                    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(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

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"))
        return dict(row), source, target
    return {}, None, None