Skip to main content
Handler return values are captured in EventResult records and can be consumed as a single value or aggregated across handlers. Repository example files:

​
Typed return values

Use the event result type to enforce return typing across handlers. 
from abxbus import BaseEvent, EventBus

class DoMathEvent(BaseEvent[int]):
    a: int
    b: int

def add(event: DoMathEvent) -> int:
    return event.a + event.b

bus = EventBus('AppBus')
bus.on(DoMathEvent, add)

event = await bus.emit(DoMathEvent(a=2, b=3))
result = await event.event_result()

​
Aggregating multiple handler results

When multiple handlers respond to the same event, collect all results. 
from abxbus import BaseEvent, EventBus

class GetConfigEvent(BaseEvent[dict]):
    pass

async def user_config(_: GetConfigEvent) -> dict:
    return {'debug': True, 'port': 8080}

async def system_config(_: GetConfigEvent) -> dict:
    return {'debug': False, 'timeout': 30}

bus = EventBus('AppBus')
bus.on(GetConfigEvent, user_config)
bus.on(GetConfigEvent, system_config)

event = await bus.emit(GetConfigEvent())
values = await event.event_results_list(raise_if_any=False, raise_if_none=False)
# [
#   {'debug': True, 'port': 8080},
#   {'debug': False, 'timeout': 30}
# ]

merged_config = {key: val for config in values for key, val in config.items()}
# {'debug': False, 'port': 8080, 'timeout': 30}

​
event_results_list / eventResultsList options

event_results_list(
    timeout: float | None = None,
    include: EventResultFilter = ...,
    raise_if_any: bool = True,
    raise_if_none: bool = True,
) -> list[Any]

# examples
await event.event_results_list(raise_if_any=False, raise_if_none=False)
await event.event_results_list(include=lambda event_result: isinstance(event_result.result, dict), raise_if_any=False)
await event.event_results_list(timeout=0.25)
  • raise_if_any: raise if any handler ended with an error.
  • raise_if_none: raise only when no handlers returned a valid value after filtering; it does not raise just because one handler returned None/null/undefined.
  • Default filtering in both runtimes includes only completed, successful, non-empty scalar/object/list values (and excludes forwarded BaseEvent returns).

​
Per-handler inspection

Both implementations keep per-handler result metadata in addition to event_results_list / eventResultsList. 
by_name = {result.handler_name: result.result for result in event.event_results.values()}