babylon.engine.adapters.aggregation_mixin

AggregationMixin for graph aggregation operations.

Extracted from inmemory_adapter to reduce class size and unify duplicate aggregation logic for nodes and edges.

Uses a unified _aggregate_items method instead of separate _aggregate_nodes and _aggregate_edges implementations.

Classes

AggregationMixin()

Mixin providing aggregation operations for graph adapters.
class babylon.engine.adapters.aggregation_mixin.AggregationMixin[source]

Bases: object

Mixin providing aggregation operations for graph adapters.

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

Example

>>> class MyAdapter(AggregationMixin):
...     def __init__(self, graph):
...         self._graph = graph
>>> # adapter.aggregate("nodes", group_by="type", agg_func="count")
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.