"""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}")
"""
from __future__ import annotations
import logging
from collections.abc import Awaitable, Callable
from inspect import getattr_static, iscoroutinefunction
from typing import TYPE_CHECKING, TypeAlias
from .models import Event
if TYPE_CHECKING:
from .models import EventType
[docs]
logger = logging.getLogger(__name__)
"""Logger for the cb_events.router module."""
[docs]
HandlerFunc: TypeAlias = Callable[[Event], Awaitable[None]]
"""Async handler signature accepted by Router decorators."""
async def _dispatch_handler(handler: HandlerFunc, event: Event) -> None:
"""Invoke a handler and log failures without stopping dispatch.
Args:
handler: Async callable that processes the event.
event: The event to pass to the handler.
"""
try:
await handler(event)
# BaseException (including CancelledError) intentionally propagates
except Exception: # pylint: disable=broad-exception-caught
logger.exception(
"Handler %s failed for event %s (type: %s)",
_handler_name(handler),
event.id,
event.type,
)
def _is_async_callable(func: object) -> bool:
"""Return whether func produces an awaitable when invoked once."""
if iscoroutinefunction(func):
return True
call_method = getattr_static(func, "__call__", None) # pyright: ignore[reportAny] # pylint: disable=line-too-long
if call_method is not None and iscoroutinefunction(call_method): # pyright: ignore[reportAny] # pylint: disable=line-too-long
return True
underlying = getattr(func, "func", None)
if callable(underlying) and underlying is not func:
return _is_async_callable(underlying)
return False
def _handler_name(handler: object) -> str:
"""Return a safe name for logging handler failures."""
seen: set[int] = set()
current = handler
while id(current) not in seen:
seen.add(id(current))
name: str | None = getattr(current, "__name__", None)
if name:
return name
wrapped = getattr(current, "__wrapped__", None)
if callable(wrapped) and wrapped is not current:
current = wrapped
continue
func_attr = getattr(current, "func", None)
if callable(func_attr) and func_attr is not current:
current = func_attr
continue
break
return type(current).__name__
[docs]
class 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.
"""
__slots__: tuple[str, ...] = ("_handlers",)
def __init__(self) -> None:
"""Initialize router with an empty handler registry."""
self._handlers: dict[EventType | None, list[HandlerFunc]] = {}
def _register(self, key: EventType | None, func: HandlerFunc) -> HandlerFunc:
"""Validate and register a handler function.
Args:
key: Event type to register for, or None for wildcard handlers.
func: Async handler function to register.
Returns:
The registered handler function unchanged.
Raises:
TypeError: If the handler is not async.
"""
if not _is_async_callable(func):
msg = f"Handler {_handler_name(func)} must be async"
raise TypeError(msg)
self._handlers.setdefault(key, []).append(func)
return func
[docs]
def on(self, event_type: EventType) -> Callable[[HandlerFunc], HandlerFunc]:
"""Register handler for a specific event type.
Args:
event_type: Event category to associate with the handler.
Returns:
Decorator that registers the handler and returns it unchanged.
"""
def decorator(func: HandlerFunc) -> HandlerFunc:
return self._register(event_type, func)
return decorator
[docs]
def on_any(self) -> Callable[[HandlerFunc], HandlerFunc]:
"""Register handler for all event types.
Returns:
Decorator that registers the handler and returns it unchanged.
"""
def decorator(func: HandlerFunc) -> HandlerFunc:
return self._register(None, func)
return decorator
[docs]
async def dispatch(self, event: 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.
Args:
event: Event instance to dispatch to registered handlers.
"""
handlers = [
*self._handlers.get(None, []),
*self._handlers.get(event.type, []),
]
if not handlers:
return
logger.debug(
"Dispatching %s event %s to %d handlers",
event.type,
event.id,
len(handlers),
)
for handler in handlers:
await _dispatch_handler(handler, event)