How to Aggregate Results from Multiple Handlers in Bubus: A Complete Guide

Bubus aggregates handler results by storing each outcome in an EventResult object attached to the event, providing helper methods like event_results_by_handler_name() and event_results_flat_dict() to collect, filter, and merge outputs after all handlers complete.

When you dispatch an event through the Bubus event bus, multiple handlers can process the same BaseEvent instance concurrently. Unlike simple pub/sub systems that discard return values, Bubus captures every handler's output in a structured EventResult object. These results live inside the event's event_results dictionary and remain accessible after the event_completed_signal fires, enabling sophisticated aggregation patterns for configuration building, data collection, and result selection.

Understanding the Aggregation Architecture

The EventResult Lifecycle

The aggregation process begins before any handler executes. In bubus/service.py (lines 66-74), the EventBus.process_event() method creates a pending EventResult for every registered handler that will process the event:

# From bubus/service.py - process_event method

event.event_result_update(
    handler_id=handler.handler_id,
    handler_name=handler.handler_name,
    status='pending'
)

This pre-registration guarantees that the event knows exactly which handlers will run. As each handler completes, its EventResult updates with the return value and a status of 'completed' (or 'error' if an exception occurred). Only after all handlers and any child events finish does the bus set the event_completed_signal, making the results safe to aggregate.

The Core Filtering Method

All aggregation helpers rely on event_results_filtered() in bubus/models.py (lines 474-527). This method:

1. Waits for the event_completed_signal to ensure all handlers finished
2. Optionally raises the first encountered error if raise_if_any=True
3. Applies a custom filter function (include) to determine which results to return

The default filter, _event_result_is_truthy, automatically excludes:

• Pending or error-filled results
• None values
• BaseEvent instances (to prevent accidental event chaining loops)

Aggregation Helper Methods

Bubus provides six primary helper methods on BaseEvent to shape aggregated results according to your application's needs.

Mapping Results by Handler Identity

When you need to identify which handler produced which result, use the mapping methods defined in bubus/models.py (lines 534-564):

• event_results_by_handler_id(): Returns dict[int, Any] mapping the numeric handler_id to the result. Use this when you need stable, unique keys that won't change if you rename functions.
• event_results_by_handler_name(): Returns dict[str, Any] mapping the string handler_name (typically the function name) to the result. This produces more readable output for debugging and logging.

Flattening Collections

For handlers that return collections you want to merge, Bubus offers flattening methods:

event_results_flat_dict() (lines 593-626 in bubus/models.py) merges all handler results (which must be dictionaries) into a single dictionary. By default, it raises a ValueError if multiple handlers return the same key (raise_if_conflicts=True), but you can set raise_if_conflicts=False to let the last handler's value win.

event_results_flat_list() performs the same operation for list-returning handlers, concatenating all lists into a single list while preserving the execution order of handlers.

Selecting Single Values

When you expect only one meaningful result across all handlers, event_result() (singular) returns the first non-None result according to the default truthy filter. This is useful for "first-wins" strategies or when you register multiple handlers but only one is expected to return a value under normal circumstances.

Practical Code Examples

Aggregating Results by Handler Name

This example demonstrates the most common aggregation pattern: collecting results into a readable dictionary keyed by function name.

from bubus import EventBus, BaseEvent

class MyEvent(BaseEvent):
    pass

bus = EventBus()

@bus.on('*')
def handler_a(event: MyEvent) -> int:
    return 1

@bus.on('*')
def handler_b(event: MyEvent) -> int:
    return 2

# Dispatch and wait for all handlers

event = await bus.dispatch(MyEvent())

# Returns {'handler_a': 1, 'handler_b': 2}

results_by_name = await event.event_results_by_handler_name()
print(results_by_name)

Merging Configuration Dictionaries

When building configuration objects where different handlers contribute specific keys, use event_results_flat_dict() to merge the partial configs.

@bus.on('*')
def cfg_part_one(_: MyEvent) -> dict:
    return {'host': 'localhost', 'port': 8000}

@bus.on('*')
def cfg_part_two(_: MyEvent) -> dict:
    return {'debug': True, 'log_level': 'info'}

event = await bus.dispatch(MyEvent())

# Merge, raising ValueError on key collisions by default

merged = await event.event_results_flat_dict()

# Result: {'host': 'localhost', 'port': 8000, 'debug': True, 'log_level': 'info'}

# Allow last-write-wins for conflicting keys

merged_no_raise = await event.event_results_flat_dict(raise_if_conflicts=False)

Concatenating Lists from Multiple Handlers

For data collection scenarios where handlers append items to a shared dataset, event_results_flat_list() preserves execution order.

@bus.on('*')
def list_a(_: MyEvent) -> list[int]:
    return [1, 2]

@bus.on('*')
def list_b(_: MyEvent) -> list[int]:
    return [3, 4]

event = await bus.dispatch(MyEvent())
flat = await event.event_results_flat_list()

# Result: [1, 2, 3, 4]

print(flat)

Selecting the First Valid Result

When using a "first-wins" strategy or when only one handler is expected to return a value, use the singular event_result() method.

@bus.on('*')
def maybe_none(_: MyEvent) -> None:
    return None

@bus.on('*')
def provides_value(_: MyEvent) -> str:
    return "final answer"

event = await bus.dispatch(MyEvent())
first = await event.event_result()

# Result: "final answer" (skips the None)

print(first)

Custom Filtering for Specific Result Types

To aggregate only results meeting specific criteria, pass a custom filter function to event_results_list().

# Only keep even numbers from handlers that return ints

def is_even(event_result):
    return isinstance(event_result.result, int) and event_result.result % 2 == 0

event = await bus.dispatch(MyEvent())
evens = await event.event_results_list(include=is_even)

# Result: [2, 4] if handlers returned 1, 2, 3, 4

print(evens)

Summary

• Bubus captures every handler's return value in an EventResult object stored in the event's event_results dictionary, created before handlers execute in bubus/service.py.
• Aggregation occurs after completion via event_results_filtered() in bubus/models.py, which waits for the event_completed_signal and applies truthy filters by default.
• Use event_results_by_handler_name() for readable dictionaries keyed by function name, or event_results_by_handler_id() for stable numeric keys.
• Merge collections with event_results_flat_dict() for configuration objects (lines 593-626 in bubus/models.py) or event_results_flat_list() for concatenated sequences.
• Select single values using event_result() when expecting only one meaningful return, or apply custom filters via the include parameter for domain-specific aggregation logic.

Frequently Asked Questions

How does Bubus handle errors when aggregating results from multiple handlers?

By default, event_results_filtered() waits for all handlers to complete before returning. If you set raise_if_any=True, it will raise the first error encountered during handler execution immediately upon aggregation. Otherwise, error results remain in the event_results dictionary with a status of 'error', and the default truthy filter excludes them from final aggregated output.

Can I aggregate results while handlers are still running?

No, aggregation helpers in bubus/models.py explicitly wait for the event_completed_signal before reading from event_results. This signal fires only after all handlers and any child events finish execution, ensuring data consistency and preventing race conditions when merging partial results.

What happens if two handlers return dictionaries with overlapping keys?

When using event_results_flat_dict() in bubus/models.py (lines 593-626), the default behavior raises a ValueError if key collisions occur between handler results. You can override this by passing raise_if_conflicts=False, which implements a last-write-wins strategy where the last executed handler's value for a given key persists in the merged dictionary.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}