babylon.engine.graph_protocol

Graph Protocol definition for backend-agnostic graph operations.

Slice 1.7: The Graph Bridge

The GraphProtocol is a typing.Protocol that defines the abstract interface for all graph operations. Systems interact with the simulation graph ONLY through this protocol. The concrete implementation (NetworkX, DuckDB) is hidden behind the adapter.

Design Principles:

Backend-agnostic: Works with NetworkX now, DuckDB later
Set-oriented: Think in tables, not just objects (DuckDB-ready)
Minimal but complete: Just enough methods to cover all System needs
Lazy evaluation: Return iterators/generators, not materialized lists

Protocol Methods (18 total):

Node CRUD: add_node, get_node, update_node, remove_node Edge CRUD: add_edge, get_edge, update_edge, remove_edge Traversal: get_neighborhood, execute_traversal, shortest_path Set Ops: query_nodes, query_edges, count_nodes, count_edges, aggregate Graph Attrs: get_graph_attr, set_graph_attr

Classes

GraphProtocol(*args, **kwargs)

Protocol for backend-agnostic graph operations.

class babylon.engine.graph_protocol.GraphProtocol(*args, **kwargs)[source]

Bases: Protocol

Protocol for backend-agnostic graph operations.

Systems interact with the simulation graph ONLY through this protocol. The concrete implementation (NetworkX, DuckDB) is hidden behind adapters.

This protocol is runtime_checkable, enabling isinstance() checks for protocol compliance.

Example

>>> class MyAdapter:
...     def add_node(self, node_id: str, node_type: str, **attrs: Any) -> None:
...         pass
...     # ... implement all 16 methods
>>> adapter = MyAdapter()
>>> isinstance(adapter, GraphProtocol)
True

add_node(node_id, node_type, **attributes)[source]

Add a node with type marker and arbitrary attributes.

Parameters:

node_id (str) – Unique identifier for the node.
node_type (str) – Discriminator for polymorphism (e.g., ‘social_class’).
**attributes (Any) – Type-specific attributes to store on the node.

Return type:

None

get_node(node_id)[source]

Retrieve node by ID.

Parameters:: node_id (str) – The node identifier to look up.
Return type:: GraphNode | None
Returns:: GraphNode model if found, None otherwise.

update_node(node_id, **attributes)[source]

Partial update of node attributes (merge, not replace).

Parameters:

node_id (str) – The node identifier to update.
**attributes (Any) – Attributes to update (merged with existing).

Raises:

KeyError – If node does not exist.

Return type:

None

remove_node(node_id)[source]

Remove node and all incident edges.

Parameters:: node_id (str) – The node identifier to remove.
Raises:: KeyError – If node does not exist.
Return type:: None

add_edge(source, target, edge_type, weight=1.0, **attributes)[source]

Add directed edge with type, weight, and attributes.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge category (e.g., ‘SOLIDARITY’, ‘EXPLOITATION’).
weight (float) – Generic weight (default 1.0).
**attributes (Any) – Type-specific attributes to store on the edge.

Return type:

None

get_edge(source, target, edge_type)[source]

Retrieve specific edge by source, target, and type.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge type to match.

Return type:

GraphEdge | None

Returns:

GraphEdge model if found, None otherwise.

update_edge(source, target, edge_type, **attributes)[source]

Partial update of edge attributes.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge type to match.
**attributes (Any) – Attributes to update (merged with existing).

Raises:

KeyError – If edge does not exist.

Return type:

None

remove_edge(source, target, edge_type)[source]

Remove specific edge.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge type to match.

Raises:

KeyError – If edge does not exist.

Return type:

None

get_neighborhood(node_id, radius=1, edge_types=None, direction='out')[source]

Get all nodes within radius hops of the source node.

Parameters:

node_id (str) – Center node for neighborhood.
radius (int) – Maximum hop distance (1 = immediate neighbors).
edge_types (set[str] | None) – Filter to specific edge types (None = all).
direction (Literal['out', 'in', 'both']) – Which edges to follow: outgoing, incoming, or both.

Return type:

Any

Returns:

SubgraphView or equivalent containing nodes in neighborhood.

Raises:

KeyError – If node does not exist.

execute_traversal(query)[source]

Execute a generic traversal query.

This is the hook for complex operations like percolation analysis, pathfinding, and component detection.

Parameters:: query (TraversalQuery) – TraversalQuery specifying the traversal.
Return type:: TraversalResult
Returns:: TraversalResult with nodes, edges, paths, or aggregates.
Raises:: ValueError – If query_type is not supported.

shortest_path(source, target, edge_types=None, weight_attr=None)[source]

Find shortest path between two nodes.

Parameters:

source (str) – Start node ID.
target (str) – End node ID.
edge_types (set[str] | None) – Filter to specific edge types.
weight_attr (str | None) – Attribute to use as weight (None = hop count).

Return type:

list[str] | None

Returns:

List of node IDs in path, or None if no path exists.

query_nodes(node_type=None, predicate=None, attributes=None)[source]

Query nodes with optional filtering.

Returns an iterator for DuckDB compatibility (lazy evaluation).

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 (DuckDB-translatable).

Return type:

Iterator[GraphNode]

Returns:

Iterator of matching GraphNode models.

query_edges(edge_type=None, predicate=None, min_weight=None, max_weight=None)[source]

Query edges with optional filtering.

Returns an iterator for DuckDB compatibility (lazy evaluation).

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.

Return type:

Iterator[GraphEdge]

Returns:

Iterator of matching GraphEdge models.

count_nodes(node_type=None)[source]

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)[source]

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.

aggregate(target, group_by=None, agg_func='count', agg_attr=None)[source]

Aggregate over nodes or edges.

Parameters:

target (Literal['nodes', 'edges']) – Whether to aggregate nodes or edges.
group_by (str | None) – Attribute to group by (e.g., ‘type’).
agg_func (Literal['count', 'sum', 'avg', 'min', 'max']) – Aggregation function to apply.
agg_attr (str | None) – Attribute to aggregate (required for sum/avg/min/max).

Return type:

dict[str, float]

Returns:

Dict mapping group keys to aggregated values.

Example

>>> graph.aggregate("nodes", group_by="type")
{"social_class": 4, "territory": 2}

get_graph_attr(key, default=None)[source]

Retrieve a graph-level attribute.

Graph attributes store global metadata (e.g., economy state, base_year, tick_dynamics). Maps to a metadata table in DuckDB.

Parameters:

key (str) – Attribute name to retrieve.
default (Any) – Value to return if attribute not present.

Return type:

Any

Returns:

The attribute value or default.

set_graph_attr(key, value)[source]

Set a graph-level attribute.

Parameters:

key (str) – Attribute name to set.
value (Any) – Value to store.

Return type:

None

__init__(*args, **kwargs)