Skip to content

graflo.architecture.schema.context.graph

Adjacency index over a schema's vertex types.

EdgeConfig is keyed only by :data:~graflo.architecture.graph_types.EdgeId, so "which edges touch this vertex type" has no answer without a scan. This module builds that index once and exposes the schema graph as a navigable object.

Naming discipline. "Neighbours" here means adjacent vertex types in the schema, never adjacent instances in a data graph. The instance-plane counterpart is Connection.graph_neighbors; the two must never share a name or an endpoint.

Attributes

Classes

SchemaGraph

Read-only adjacency index over a :class:Schema's vertex types.

Built once per schema and never mutates it. Plain dicts throughout — no networkx, because this is layer 2 and the whole point is to stay free of heavyweight dependencies.

Source code in graflo/architecture/schema/context/graph.py 
 74
 75
 76
 77
 78
 79
 80
 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
120
121
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
160
161
162
163
164
165
166
167
168
169
170
171
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
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
class SchemaGraph:
    """Read-only adjacency index over a :class:`Schema`'s vertex types.

    Built once per schema and never mutates it. Plain dicts throughout — no
    networkx, because this is layer 2 and the whole point is to stay free of
    heavyweight dependencies.
    """

    def __init__(self, schema: Schema) -> None:
        self._schema = schema
        core = schema.core_schema
        self._vertex_types = frozenset(core.vertex_config.vertex_set)
        self._out: dict[str, list[EdgeId]] = {name: [] for name in self._vertex_types}
        self._in: dict[str, list[EdgeId]] = {name: [] for name in self._vertex_types}
        self._edges: dict[EdgeId, Edge] = {}

        for edge in core.edge_config.edges:
            edge_id = edge.edge_id
            self._edges[edge_id] = edge
            source, target, _relation = edge_id
            if source in self._out:
                self._out[source].append(edge_id)
            if target in self._in:
                self._in[target].append(edge_id)

        for adjacency in (self._out, self._in):
            for edge_ids in adjacency.values():
                edge_ids.sort(key=edge_sort_key)

    @classmethod
    def from_schema(cls, schema: Schema) -> SchemaGraph:
        """Build an index for *schema*."""
        return cls(schema)

    @property
    def schema(self) -> Schema:
        """The indexed schema. Treat as read-only."""
        return self._schema

    @property
    def vertex_types(self) -> frozenset[str]:
        """Every declared vertex type name."""
        return self._vertex_types

    @property
    def edge_ids(self) -> list[EdgeId]:
        """Every declared edge id, in deterministic order."""
        return sorted(self._edges, key=edge_sort_key)

    def edge(self, edge_id: EdgeId) -> Edge:
        """Return the declared edge for *edge_id*."""
        return self._edges[edge_id]

    def out_edges(self, vertex_type: str) -> list[EdgeId]:
        """Edges whose source is *vertex_type*."""
        return list(self._out.get(vertex_type, []))

    def in_edges(self, vertex_type: str) -> list[EdgeId]:
        """Edges whose target is *vertex_type*."""
        return list(self._in.get(vertex_type, []))

    def degree(self, vertex_type: str) -> int:
        """Total incident edge count (out + in), counting self-loops twice."""
        return len(self._out.get(vertex_type, [])) + len(self._in.get(vertex_type, []))

    def isolated_types(self) -> list[str]:
        """Vertex types with no incident edge at all."""
        return sorted(name for name in self._vertex_types if self.degree(name) == 0)

    def relation_vocabulary(self) -> list[str]:
        """Distinct relation names a query over this schema may use.

        The names that label declared edges, plus the declared inverse of each:
        an inverse is a valid name for reading its edge from the target, whether
        it is stored, maintained by the database, or only declared.
        """
        edge_config = self._schema.core_schema.edge_config
        stored = {
            relation
            for _source, _target, relation in self._edges
            if relation is not None
        }
        inverses = {edge_config.inverse_of(relation) for relation in stored}
        return sorted(stored | {name for name in inverses if name is not None})

    def _traversable(
        self,
        edge_id: EdgeId,
        anchor: str,
        direction: EdgeDirection,
    ) -> str | None:
        """Return the far endpoint when *edge_id* may be followed from *anchor*.

        An edge declared ``directed=False`` is traversable both ways regardless of
        the requested direction — the same rule
        :func:`~graflo.db.edge_direction_support.default_direction_for_edge`
        applies on the instance plane.
        """
        source, target, _relation = edge_id
        undirected = not self._edges[edge_id].directed
        effective = EdgeDirection.ANY if undirected else direction

        forward = source == anchor and effective in (
            EdgeDirection.OUT,
            EdgeDirection.ANY,
        )
        backward = target == anchor and effective in (
            EdgeDirection.IN,
            EdgeDirection.ANY,
        )
        if forward:
            return target
        if backward:
            return source
        return None

    def _incident(self, vertex_type: str) -> list[EdgeId]:
        """Every edge touching *vertex_type*, deduplicated (self-loops appear once)."""
        seen: set[EdgeId] = set()
        incident: list[EdgeId] = []
        for edge_id in self._out.get(vertex_type, []) + self._in.get(vertex_type, []):
            if edge_id in seen:
                continue
            seen.add(edge_id)
            incident.append(edge_id)
        return sorted(incident, key=edge_sort_key)

    def schema_neighbors(
        self,
        vertex_type: str,
        *,
        hops: int = 1,
        direction: EdgeDirection = EdgeDirection.ANY,
        edge_relations: set[str | None] | None = None,
        edge_ids: set[EdgeId] | None = None,
    ) -> SchemaNeighborhood:
        """Vertex types adjacent to *vertex_type* within *hops*.

        Args:
            vertex_type: Seed vertex type. Must be declared.
            hops: Maximum hop distance. ``0`` returns just the seed.
            direction: Orientation followed from each frontier vertex. Defaults to
                :attr:`EdgeDirection.ANY` — deliberately unlike
                ``Connection.fetch_edges``, which defaults to ``OUT``. "What is
                adjacent to ``person`` in the schema" almost never means "only
                where person is the source"; an agent asking that wants the whole
                local shape. Edges declared ``directed=False`` are followed both
                ways whatever is requested here.
            edge_relations: Restrict traversal to these relation names (``None`` is
                a valid member, matching edges with no relation).
            edge_ids: Restrict traversal to these exact edges. Finer-grained than
                *edge_relations*, which cannot separate two dyads sharing one
                relation name. Both filters apply when both are given.

        Returns:
            SchemaNeighborhood: distances per reachable type and the edges used.

        Raises:
            KeyError: if *vertex_type* is not declared in the schema.
        """
        if vertex_type not in self._vertex_types:
            raise KeyError(
                f"Unknown vertex type {vertex_type!r}; declared: {sorted(self._vertex_types)}"
            )
        if hops < 0:
            raise ValueError(f"hops must be >= 0, got {hops}")

        distances: dict[str, int] = {vertex_type: 0}
        used: set[EdgeId] = set()
        frontier: deque[tuple[str, int]] = deque([(vertex_type, 0)])

        while frontier:
            current, depth = frontier.popleft()
            if depth >= hops:
                continue
            for edge_id in self._incident(current):
                if edge_relations is not None and edge_id[2] not in edge_relations:
                    continue
                if edge_ids is not None and edge_id not in edge_ids:
                    continue
                far = self._traversable(edge_id, current, direction)
                if far is None:
                    continue
                used.add(edge_id)
                if far not in distances:
                    distances[far] = depth + 1
                    frontier.append((far, depth + 1))

        return SchemaNeighborhood(
            origin=vertex_type,
            hops=hops,
            direction=direction,
            distances=distances,
            edges=sorted(used, key=edge_sort_key),
        )

    def relations_between(
        self,
        a: str,
        b: str,
        *,
        max_len: int = 3,
        max_paths: int = 20,
        direction: EdgeDirection = EdgeDirection.ANY,
    ) -> list[SchemaPath]:
        """Simple paths from vertex type *a* to *b*, shortest first.

        Bounded breadth-first enumeration: no vertex repeats within a path, so
        cycles terminate. Results are ordered by ``(length, edge ids)`` and are
        therefore reproducible run to run.

        Args:
            a: Source vertex type.
            b: Target vertex type.
            max_len: Maximum hops per path.
            max_paths: Maximum number of paths returned.
            direction: Orientation followed from each frontier vertex.

        Returns:
            list[SchemaPath]: paths found, possibly empty.

        Raises:
            KeyError: if either endpoint is not declared in the schema.
        """
        for name in (a, b):
            if name not in self._vertex_types:
                raise KeyError(
                    f"Unknown vertex type {name!r}; declared: {sorted(self._vertex_types)}"
                )
        if max_len < 1 or max_paths < 1:
            return []

        found: list[SchemaPath] = []
        queue: deque[tuple[str, list[str], list[EdgeId]]] = deque([(a, [a], [])])

        while queue and len(found) < max_paths:
            current, vertices, edges = queue.popleft()
            if len(edges) >= max_len:
                continue
            for edge_id in self._incident(current):
                far = self._traversable(edge_id, current, direction)
                if far is None:
                    continue
                # Paths stay simple, except that reaching the target closes the
                # walk — which is what makes ``relations_between(a, a)`` return
                # self-loops and cycles rather than nothing.
                if far in vertices and far != b:
                    continue
                next_vertices = [*vertices, far]
                next_edges = [*edges, edge_id]
                if far == b:
                    found.append(SchemaPath(vertices=next_vertices, edges=next_edges))
                    if len(found) >= max_paths:
                        break
                else:
                    queue.append((far, next_vertices, next_edges))

        found.sort(
            key=lambda path: (path.length, [edge_sort_key(e) for e in path.edges])
        )
        return found[:max_paths]

Attributes

edge_ids property

Every declared edge id, in deterministic order.

schema property

The indexed schema. Treat as read-only.

vertex_types property

Every declared vertex type name.

Methods:

__init__(schema)
Source code in graflo/architecture/schema/context/graph.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(self, schema: Schema) -> None:
    self._schema = schema
    core = schema.core_schema
    self._vertex_types = frozenset(core.vertex_config.vertex_set)
    self._out: dict[str, list[EdgeId]] = {name: [] for name in self._vertex_types}
    self._in: dict[str, list[EdgeId]] = {name: [] for name in self._vertex_types}
    self._edges: dict[EdgeId, Edge] = {}

    for edge in core.edge_config.edges:
        edge_id = edge.edge_id
        self._edges[edge_id] = edge
        source, target, _relation = edge_id
        if source in self._out:
            self._out[source].append(edge_id)
        if target in self._in:
            self._in[target].append(edge_id)

    for adjacency in (self._out, self._in):
        for edge_ids in adjacency.values():
            edge_ids.sort(key=edge_sort_key)
degree(vertex_type)

Total incident edge count (out + in), counting self-loops twice.

Source code in graflo/architecture/schema/context/graph.py 
135
136
137
def degree(self, vertex_type: str) -> int:
    """Total incident edge count (out + in), counting self-loops twice."""
    return len(self._out.get(vertex_type, [])) + len(self._in.get(vertex_type, []))
edge(edge_id)

Return the declared edge for edge_id.

Source code in graflo/architecture/schema/context/graph.py 
123
124
125
def edge(self, edge_id: EdgeId) -> Edge:
    """Return the declared edge for *edge_id*."""
    return self._edges[edge_id]
from_schema(schema) classmethod

Build an index for schema.

Source code in graflo/architecture/schema/context/graph.py 
103
104
105
106
@classmethod
def from_schema(cls, schema: Schema) -> SchemaGraph:
    """Build an index for *schema*."""
    return cls(schema)
in_edges(vertex_type)

Edges whose target is vertex_type.

Source code in graflo/architecture/schema/context/graph.py 
131
132
133
def in_edges(self, vertex_type: str) -> list[EdgeId]:
    """Edges whose target is *vertex_type*."""
    return list(self._in.get(vertex_type, []))
isolated_types()

Vertex types with no incident edge at all.

Source code in graflo/architecture/schema/context/graph.py 
139
140
141
def isolated_types(self) -> list[str]:
    """Vertex types with no incident edge at all."""
    return sorted(name for name in self._vertex_types if self.degree(name) == 0)
out_edges(vertex_type)

Edges whose source is vertex_type.

Source code in graflo/architecture/schema/context/graph.py 
127
128
129
def out_edges(self, vertex_type: str) -> list[EdgeId]:
    """Edges whose source is *vertex_type*."""
    return list(self._out.get(vertex_type, []))
relation_vocabulary()

Distinct relation names a query over this schema may use.

The names that label declared edges, plus the declared inverse of each: an inverse is a valid name for reading its edge from the target, whether it is stored, maintained by the database, or only declared.

Source code in graflo/architecture/schema/context/graph.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def relation_vocabulary(self) -> list[str]:
    """Distinct relation names a query over this schema may use.

    The names that label declared edges, plus the declared inverse of each:
    an inverse is a valid name for reading its edge from the target, whether
    it is stored, maintained by the database, or only declared.
    """
    edge_config = self._schema.core_schema.edge_config
    stored = {
        relation
        for _source, _target, relation in self._edges
        if relation is not None
    }
    inverses = {edge_config.inverse_of(relation) for relation in stored}
    return sorted(stored | {name for name in inverses if name is not None})
relations_between(a, b, *, max_len=3, max_paths=20, direction=EdgeDirection.ANY)

Simple paths from vertex type a to b, shortest first.

Bounded breadth-first enumeration: no vertex repeats within a path, so cycles terminate. Results are ordered by (length, edge ids) and are therefore reproducible run to run.

Parameters:

Name Type Description Default
a str

Source vertex type.

 required
b str

Target vertex type.

 required
max_len int

Maximum hops per path.

 3
max_paths int

Maximum number of paths returned.

 20
direction EdgeDirection

Orientation followed from each frontier vertex.

 ANY

Returns:

Type Description
list[SchemaPath]

list[SchemaPath]: paths found, possibly empty.

Raises:

Type Description
KeyError

if either endpoint is not declared in the schema.
Source code in graflo/architecture/schema/context/graph.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
def relations_between(
    self,
    a: str,
    b: str,
    *,
    max_len: int = 3,
    max_paths: int = 20,
    direction: EdgeDirection = EdgeDirection.ANY,
) -> list[SchemaPath]:
    """Simple paths from vertex type *a* to *b*, shortest first.

    Bounded breadth-first enumeration: no vertex repeats within a path, so
    cycles terminate. Results are ordered by ``(length, edge ids)`` and are
    therefore reproducible run to run.

    Args:
        a: Source vertex type.
        b: Target vertex type.
        max_len: Maximum hops per path.
        max_paths: Maximum number of paths returned.
        direction: Orientation followed from each frontier vertex.

    Returns:
        list[SchemaPath]: paths found, possibly empty.

    Raises:
        KeyError: if either endpoint is not declared in the schema.
    """
    for name in (a, b):
        if name not in self._vertex_types:
            raise KeyError(
                f"Unknown vertex type {name!r}; declared: {sorted(self._vertex_types)}"
            )
    if max_len < 1 or max_paths < 1:
        return []

    found: list[SchemaPath] = []
    queue: deque[tuple[str, list[str], list[EdgeId]]] = deque([(a, [a], [])])

    while queue and len(found) < max_paths:
        current, vertices, edges = queue.popleft()
        if len(edges) >= max_len:
            continue
        for edge_id in self._incident(current):
            far = self._traversable(edge_id, current, direction)
            if far is None:
                continue
            # Paths stay simple, except that reaching the target closes the
            # walk — which is what makes ``relations_between(a, a)`` return
            # self-loops and cycles rather than nothing.
            if far in vertices and far != b:
                continue
            next_vertices = [*vertices, far]
            next_edges = [*edges, edge_id]
            if far == b:
                found.append(SchemaPath(vertices=next_vertices, edges=next_edges))
                if len(found) >= max_paths:
                    break
            else:
                queue.append((far, next_vertices, next_edges))

    found.sort(
        key=lambda path: (path.length, [edge_sort_key(e) for e in path.edges])
    )
    return found[:max_paths]
schema_neighbors(vertex_type, *, hops=1, direction=EdgeDirection.ANY, edge_relations=None, edge_ids=None)

Vertex types adjacent to vertex_type within hops.

Parameters:

Name Type Description Default
vertex_type str

Seed vertex type. Must be declared.

 required
hops int

Maximum hop distance. 0 returns just the seed.

 1
direction EdgeDirection

Orientation followed from each frontier vertex. Defaults to :attr:EdgeDirection.ANY — deliberately unlike Connection.fetch_edges, which defaults to OUT. "What is adjacent to person in the schema" almost never means "only where person is the source"; an agent asking that wants the whole local shape. Edges declared directed=False are followed both ways whatever is requested here.

 ANY
edge_relations set[str | None] | None

Restrict traversal to these relation names (None is a valid member, matching edges with no relation).

 None
edge_ids set[EdgeId] | None

Restrict traversal to these exact edges. Finer-grained than edge_relations, which cannot separate two dyads sharing one relation name. Both filters apply when both are given.

 None

Returns:

Name Type Description
SchemaNeighborhood SchemaNeighborhood

distances per reachable type and the edges used.

Raises:

Type Description
KeyError

if vertex_type is not declared in the schema.
Source code in graflo/architecture/schema/context/graph.py 
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
def schema_neighbors(
    self,
    vertex_type: str,
    *,
    hops: int = 1,
    direction: EdgeDirection = EdgeDirection.ANY,
    edge_relations: set[str | None] | None = None,
    edge_ids: set[EdgeId] | None = None,
) -> SchemaNeighborhood:
    """Vertex types adjacent to *vertex_type* within *hops*.

    Args:
        vertex_type: Seed vertex type. Must be declared.
        hops: Maximum hop distance. ``0`` returns just the seed.
        direction: Orientation followed from each frontier vertex. Defaults to
            :attr:`EdgeDirection.ANY` — deliberately unlike
            ``Connection.fetch_edges``, which defaults to ``OUT``. "What is
            adjacent to ``person`` in the schema" almost never means "only
            where person is the source"; an agent asking that wants the whole
            local shape. Edges declared ``directed=False`` are followed both
            ways whatever is requested here.
        edge_relations: Restrict traversal to these relation names (``None`` is
            a valid member, matching edges with no relation).
        edge_ids: Restrict traversal to these exact edges. Finer-grained than
            *edge_relations*, which cannot separate two dyads sharing one
            relation name. Both filters apply when both are given.

    Returns:
        SchemaNeighborhood: distances per reachable type and the edges used.

    Raises:
        KeyError: if *vertex_type* is not declared in the schema.
    """
    if vertex_type not in self._vertex_types:
        raise KeyError(
            f"Unknown vertex type {vertex_type!r}; declared: {sorted(self._vertex_types)}"
        )
    if hops < 0:
        raise ValueError(f"hops must be >= 0, got {hops}")

    distances: dict[str, int] = {vertex_type: 0}
    used: set[EdgeId] = set()
    frontier: deque[tuple[str, int]] = deque([(vertex_type, 0)])

    while frontier:
        current, depth = frontier.popleft()
        if depth >= hops:
            continue
        for edge_id in self._incident(current):
            if edge_relations is not None and edge_id[2] not in edge_relations:
                continue
            if edge_ids is not None and edge_id not in edge_ids:
                continue
            far = self._traversable(edge_id, current, direction)
            if far is None:
                continue
            used.add(edge_id)
            if far not in distances:
                distances[far] = depth + 1
                frontier.append((far, depth + 1))

    return SchemaNeighborhood(
        origin=vertex_type,
        hops=hops,
        direction=direction,
        distances=distances,
        edges=sorted(used, key=edge_sort_key),
    )

SchemaNeighborhood

Bases: ConfigBaseModel

Vertex types reachable from a seed within a hop bound.

Source code in graflo/architecture/schema/context/graph.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class SchemaNeighborhood(ConfigBaseModel):
    """Vertex types reachable from a seed within a hop bound."""

    origin: str = PydanticField(..., description="Vertex type the walk started from.")
    hops: int = PydanticField(..., description="Hop bound the walk honoured.")
    direction: EdgeDirection = PydanticField(
        ..., description="Orientation followed from each frontier vertex."
    )
    distances: dict[str, int] = PydanticField(
        ...,
        description="Reachable vertex type -> hop distance from origin (origin itself is 0).",
    )
    edges: list[EdgeId] = PydanticField(
        ..., description="Edges traversed to reach the neighbourhood, deduplicated."
    )

    @property
    def vertex_types(self) -> list[str]:
        """Reachable vertex types, nearest first then alphabetical."""
        return sorted(self.distances, key=lambda name: (self.distances[name], name))

Attributes

direction = PydanticField(..., description='Orientation followed from each frontier vertex.') class-attribute instance-attribute
distances = PydanticField(..., description='Reachable vertex type -> hop distance from origin (origin itself is 0).') class-attribute instance-attribute
edges = PydanticField(..., description='Edges traversed to reach the neighbourhood, deduplicated.') class-attribute instance-attribute
hops = PydanticField(..., description='Hop bound the walk honoured.') class-attribute instance-attribute
origin = PydanticField(..., description='Vertex type the walk started from.') class-attribute instance-attribute
vertex_types property

Reachable vertex types, nearest first then alphabetical.

SchemaPath

Bases: ConfigBaseModel

One path between two vertex types, as an alternating vertex/edge walk.

Source code in graflo/architecture/schema/context/graph.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class SchemaPath(ConfigBaseModel):
    """One path between two vertex types, as an alternating vertex/edge walk."""

    vertices: list[str] = PydanticField(
        ..., description="Vertex types visited, from source to target inclusive."
    )
    edges: list[EdgeId] = PydanticField(
        ..., description="Edges traversed, one fewer than ``vertices``."
    )

    @property
    def length(self) -> int:
        """Number of hops (edges) in this path."""
        return len(self.edges)

Attributes

edges = PydanticField(..., description='Edges traversed, one fewer than ``vertices``.') class-attribute instance-attribute
length property

Number of hops (edges) in this path.

vertices = PydanticField(..., description='Vertex types visited, from source to target inclusive.') class-attribute instance-attribute

Functions:

edge_sort_key(edge_id)

Total order over edge ids.

relation is str | None, so plain tuple comparison raises TypeError as soon as a relation-less edge meets a named one. Every ordering in this package goes through here so results stay deterministic and comparable.

Source code in graflo/architecture/schema/context/graph.py 
25
26
27
28
29
30
31
32
33
def edge_sort_key(edge_id: EdgeId) -> tuple[str, str, str]:
    """Total order over edge ids.

    ``relation`` is ``str | None``, so plain tuple comparison raises ``TypeError``
    as soon as a relation-less edge meets a named one. Every ordering in this
    package goes through here so results stay deterministic and comparable.
    """
    source, target, relation = edge_id
    return (source, target, relation or "")

neighborhood_distances(graph, seeds, *, hops=1, direction=EdgeDirection.ANY, edge_relations=None, edge_ids=None)

Hop distance to every type reachable from any seed, nearest seed winning.

The multi-seed union of :meth:SchemaGraph.schema_neighbors. Both consumers of "what is near what I asked about" — rank-then-budget schema context and project_manifest's depth — go through here, so the two cannot drift apart on direction handling, undirected edges or the treatment of hops=0.

Parameters:

Name Type Description Default
graph SchemaGraph

Adjacency index to walk.

 required
seeds Iterable[str]

Vertex types to start from. Each must be declared.

 required
hops int

Maximum hop distance. 0 yields just the seeds.

 1
direction EdgeDirection

Orientation followed from each frontier vertex.

 ANY
edge_relations set[str | None] | None

Restrict traversal to these relation names.

 None
edge_ids set[EdgeId] | None

Restrict traversal to these exact edges.

 None

Returns:

Name Type Description
dict dict[str, int]

reachable vertex type -> distance from the nearest seed (seeds map
dict[str, int]

to 0). Empty when seeds is empty.

Raises:

Type Description
KeyError

if a seed is not declared in the schema.
Source code in graflo/architecture/schema/context/graph.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
def neighborhood_distances(
    graph: SchemaGraph,
    seeds: Iterable[str],
    *,
    hops: int = 1,
    direction: EdgeDirection = EdgeDirection.ANY,
    edge_relations: set[str | None] | None = None,
    edge_ids: set[EdgeId] | None = None,
) -> dict[str, int]:
    """Hop distance to every type reachable from *any* seed, nearest seed winning.

    The multi-seed union of :meth:`SchemaGraph.schema_neighbors`. Both consumers of
    "what is near what I asked about" — rank-then-budget schema context and
    ``project_manifest``'s ``depth`` — go through here, so the two cannot drift
    apart on direction handling, undirected edges or the treatment of ``hops=0``.

    Args:
        graph: Adjacency index to walk.
        seeds: Vertex types to start from. Each must be declared.
        hops: Maximum hop distance. ``0`` yields just the seeds.
        direction: Orientation followed from each frontier vertex.
        edge_relations: Restrict traversal to these relation names.
        edge_ids: Restrict traversal to these exact edges.

    Returns:
        dict: reachable vertex type -> distance from the nearest seed (seeds map
        to ``0``). Empty when *seeds* is empty.

    Raises:
        KeyError: if a seed is not declared in the schema.
    """
    distances: dict[str, int] = {}
    for seed in seeds:
        neighborhood = graph.schema_neighbors(
            seed,
            hops=hops,
            direction=direction,
            edge_relations=edge_relations,
            edge_ids=edge_ids,
        )
        for name, distance in neighborhood.distances.items():
            if name not in distances or distance < distances[name]:
                distances[name] = distance
    return distances