babylon.engine.adapters.query_mixin

QueryMixin for graph query operations.

Extracted from inmemory_adapter to reduce class size. Provides node and edge querying and counting functionality.

Classes

QueryMixin()

Mixin providing query operations for graph adapters.

class babylon.engine.adapters.query_mixin.QueryMixin

Bases: object

Mixin providing query operations for graph adapters.

Requires the class using this mixin to have a _graph attribute that is a NetworkX DiGraph.

Example

>>> class MyAdapter(QueryMixin):
...     def __init__(self, graph):
...         self._graph = graph
>>> # adapter.query_nodes(node_type="social_class")

query_nodes(node_type=None, predicate=None, attributes=None)

Query nodes with optional filtering.

Parameters:

• node_type (str | None) – Filter by node type (None = all types).

• predicate (Callable[[GraphNode], bool] | None) – Python callable for complex filtering.

• attributes (dict[str, Any] | None) – Attribute equality filter.

Yields:

Matching GraphNode models.

Return type:

Iterator[GraphNode]

query_edges(edge_type=None, predicate=None, min_weight=None, max_weight=None)

Query edges with optional filtering.

Parameters:

• edge_type (str | None) – Filter by edge type.

• predicate (Callable[[GraphEdge], bool] | None) – Python callable for complex filtering.

• min_weight (float | None) – Minimum weight threshold.

• max_weight (float | None) – Maximum weight threshold.

Yields:

Matching GraphEdge models.

Return type:

Iterator[GraphEdge]

count_nodes(node_type=None)

Count nodes, optionally by type.

Parameters:

node_type (str | None) – Filter by node type (None = count all).

Return type:

int

Returns:

Number of matching nodes.

count_edges(edge_type=None)

Count edges, optionally by type.

Parameters:

edge_type (str | None) – Filter by edge type (None = count all).

Return type:

int

Returns:

Number of matching edges.