cb_events.router

Event routing with decorator-based handler registration.

This module provides the Router class for dispatching events to registered async handlers based on event type.

Module Attributes:

HandlerFunc: Type alias for async handler functions accepting an Event

and returning None.

Example

Basic router setup:

from cb_events import Router, EventType, Event

router = Router()

@router.on(EventType.TIP)
async def handle_tip(event: Event) -> None:
    print(f"Tip received: {event.id}")

@router.on_any()
async def log_all(event: Event) -> None:
    print(f"Event: {event.type}")

Attributes

HandlerFunc	Async handler signature accepted by Router decorators.
logger	Logger for the cb_events.router module.

Classes

Router

Routes events to registered async handlers.

Module Contents

class cb_events.router.Router

Routes events to registered async handlers.

Provides decorator-based registration for event handlers with support for both type-specific and wildcard (catch-all) handlers. Handlers execute sequentially in registration order; errors are logged but do not prevent subsequent handlers from running.

Example

Register and dispatch handlers:

router = Router()

@router.on(EventType.TIP)
async def handle_tip(event: Event) -> None:
    print(f"Tip: {event.tip.tokens if event.tip else 0}")

@router.on_any()
async def log_event(event: Event) -> None:
    print(f"Event received: {event.type}")

await router.dispatch(event)

Note

Wildcard handlers (registered via on_any()) execute before type-specific handlers for each event to support logging or preprocessing regardless of event type.

Initialize router with an empty handler registry.

async dispatch(event: cb_events.models.Event) → None

Dispatch an event to all matching handlers.

Executes wildcard handlers first, then type-specific handlers, in registration order. Handler exceptions are caught, logged, and do not propagate or prevent other handlers from executing.

Parameters:

event – Event instance to dispatch to registered handlers.

on(event_type: cb_events.models.EventType) → collections.abc.Callable[[HandlerFunc], HandlerFunc]

Register handler for a specific event type.

Parameters:

event_type – Event category to associate with the handler.

Returns:

Decorator that registers the handler and returns it unchanged.

on_any() → collections.abc.Callable[[HandlerFunc], HandlerFunc]

Register handler for all event types.

Returns:

Decorator that registers the handler and returns it unchanged.

type cb_events.router.HandlerFunc = Callable[[Event], Awaitable[None]]

Async handler signature accepted by Router decorators.

cb_events.router.logger

Logger for the cb_events.router module.