babylon.engine.adapters.subgraph_filter

SubgraphFilterBuilder for constructing filtered subgraphs.

Extracted from inmemory_adapter._build_filtered_subgraph to reduce cyclomatic complexity and improve testability.

Uses the Builder pattern to chain filter operations.

Classes

SubgraphFilterBuilder(graph)

Builds filtered subgraphs from NetworkX DiGraphs.

class babylon.engine.adapters.subgraph_filter.SubgraphFilterBuilder(graph)

Bases: object

Builds filtered subgraphs from NetworkX DiGraphs.

Uses the Builder pattern to apply node and edge filters incrementally. Each filter method returns self for method chaining.

Example

>>> builder = SubgraphFilterBuilder(graph)
>>> subgraph = (
...     builder
...     .with_nodes({"C001", "C002"})
...     .with_node_types({"social_class"})
...     .with_edge_types({"SOLIDARITY"})
...     .build()
... )

Parameters:

graph (nx.DiGraph[str])

__init__(graph)

Initialize builder with source graph.

Parameters:

graph – The NetworkX DiGraph to filter.

with_nodes(nodes)

Filter to specific node IDs.

Parameters:

nodes (set[str] | None) – Set of node IDs to include, or None for all nodes.

Return type:

SubgraphFilterBuilder

Returns:

Self for method chaining.

with_node_types(node_types)

Filter nodes by type.

Parameters:

node_types (set[str] | None) – Set of node types to include, or None for all types.

Return type:

SubgraphFilterBuilder

Returns:

Self for method chaining.

with_edge_types(edge_types)

Filter edges by type.

Parameters:

edge_types (set[str] | None) – Set of edge types to include, or None for all types.

Return type:

SubgraphFilterBuilder

Returns:

Self for method chaining.

with_weight_range(min_weight=None, max_weight=None)

Filter edges by weight range.

Parameters:

• min_weight (float | None) – Minimum edge weight (inclusive), or None for no minimum.

• max_weight (float | None) – Maximum edge weight (inclusive), or None for no maximum.

Return type:

SubgraphFilterBuilder

Returns:

Self for method chaining.

from_query(query, include_all_nodes=False)

Configure builder from a TraversalQuery.

Parameters:

• query (TraversalQuery) – TraversalQuery with filters to apply.

• include_all_nodes (bool) – If True, include all nodes regardless of start_nodes. Used for BFS/DFS where start_nodes are starting points, not filters.

Return type:

SubgraphFilterBuilder

Returns:

Self for method chaining.

build()

Build the filtered subgraph.

Applies all configured filters and returns a new DiGraph.

Returns:

A new NetworkX DiGraph containing only filtered nodes and edges.

Return type:

nx.DiGraph[str]