Skip to content

Events

EventDispatcher

Dispatches events to registered handlers.

Features: - Multiple handlers per event type - Error isolation: a failing handler does not affect others - off() to unsubscribe - Decorator mode: @dispatcher.on("a")

Source code in bidiwave/events/dispatcher.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class EventDispatcher:
    """Dispatches events to registered handlers.

    Features:
    - Multiple handlers per event type
    - Error isolation: a failing handler does not affect others
    - off() to unsubscribe
    - Decorator mode: @dispatcher.on("a")
    """

    def __init__(self) -> None:
        self._handlers: dict[str, list[AsyncHandler]] = {}

    @overload
    def on(self, event_type: str, handler: AsyncHandler) -> Subscription: ...
    @overload
    def on(
        self, event_type: str, handler: None = None
    ) -> Callable[[AsyncHandler], AsyncHandler]: ...

    def on(
        self,
        event_type: str,
        handler: AsyncHandler | None = None,
    ) -> Subscription | Callable[[AsyncHandler], AsyncHandler]:
        """Registers a handler for an event type.

        If handler is None, acts as a decorator.
        If handler is provided, registers and returns a Subscription.

        Args:
            event_type: Event type (e.g: "log.entryAdded").
            handler: Async function that receives the event.

        Returns:
            Subscription if handler is provided, decorator if not.
        """
        if handler is not None:
            if event_type not in self._handlers:
                self._handlers[event_type] = []
            self._handlers[event_type].append(handler)
            logger.debug(
                "Subscribed to %s (total: %d)",
                event_type,
                len(self._handlers[event_type]),
            )
            return Subscription(event_type, handler, self)

        def decorator(func: AsyncHandler) -> AsyncHandler:
            if event_type not in self._handlers:
                self._handlers[event_type] = []
            self._handlers[event_type].append(func)
            logger.debug(
                "Subscribed to %s (total: %d)",
                event_type,
                len(self._handlers[event_type]),
            )
            return func

        return decorator

    def off(self, subscription: Subscription) -> None:
        """Unsubscribes a handler."""
        handlers = self._handlers.get(subscription.event_type, [])
        if subscription.handler in handlers:
            handlers.remove(subscription.handler)
            logger.debug(
                "Unsubscribed from %s (remaining: %d)",
                subscription.event_type,
                len(handlers),
            )
            if not handlers:
                del self._handlers[subscription.event_type]

    async def dispatch(self, method: str, params: dict[str, Any]) -> None:
        """Dispatches an event to all subscribed handlers.

        If a handler fails, the error is logged but not propagated
        (error isolation).
        """
        try:
            event = parse_event(method, params)
        except Exception as e:
            logger.warning(
                "Failed to parse event %s: %s: %s — passing raw params",
                method,
                type(e).__name__,
                e,
            )
            event = params  # type: ignore[assignment]
        handlers = list(self._handlers.get(method, []))

        for handler in handlers:
            try:
                await handler(event)
            except Exception as e:
                logger.error(
                    "Handler error for %s: %s: %s",
                    method,
                    type(e).__name__,
                    e,
                    exc_info=True,
                )

on 

on(event_type: str, handler: AsyncHandler) -> Subscription

on(event_type: str, handler: None = None) -> Callable[[AsyncHandler], AsyncHandler]

on(event_type: str, handler: AsyncHandler | None = None) -> Subscription | Callable[[AsyncHandler], AsyncHandler]

Registers a handler for an event type.

If handler is None, acts as a decorator. If handler is provided, registers and returns a Subscription.

Parameters:

Name Type Description Default
event_type str

Event type (e.g: "log.entryAdded").

 required
handler AsyncHandler | None

Async function that receives the event.

 None

Returns:

Type Description
Subscription | Callable[[AsyncHandler], AsyncHandler]

Subscription if handler is provided, decorator if not.
Source code in bidiwave/events/dispatcher.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def on(
    self,
    event_type: str,
    handler: AsyncHandler | None = None,
) -> Subscription | Callable[[AsyncHandler], AsyncHandler]:
    """Registers a handler for an event type.

    If handler is None, acts as a decorator.
    If handler is provided, registers and returns a Subscription.

    Args:
        event_type: Event type (e.g: "log.entryAdded").
        handler: Async function that receives the event.

    Returns:
        Subscription if handler is provided, decorator if not.
    """
    if handler is not None:
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)
        logger.debug(
            "Subscribed to %s (total: %d)",
            event_type,
            len(self._handlers[event_type]),
        )
        return Subscription(event_type, handler, self)

    def decorator(func: AsyncHandler) -> AsyncHandler:
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(func)
        logger.debug(
            "Subscribed to %s (total: %d)",
            event_type,
            len(self._handlers[event_type]),
        )
        return func

    return decorator

off 

off(subscription: Subscription) -> None

Unsubscribes a handler.

Source code in bidiwave/events/dispatcher.py 
74
75
76
77
78
79
80
81
82
83
84
85
def off(self, subscription: Subscription) -> None:
    """Unsubscribes a handler."""
    handlers = self._handlers.get(subscription.event_type, [])
    if subscription.handler in handlers:
        handlers.remove(subscription.handler)
        logger.debug(
            "Unsubscribed from %s (remaining: %d)",
            subscription.event_type,
            len(handlers),
        )
        if not handlers:
            del self._handlers[subscription.event_type]

dispatch async 

dispatch(method: str, params: dict[str, Any]) -> None

Dispatches an event to all subscribed handlers.

If a handler fails, the error is logged but not propagated (error isolation).

Source code in bidiwave/events/dispatcher.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def dispatch(self, method: str, params: dict[str, Any]) -> None:
    """Dispatches an event to all subscribed handlers.

    If a handler fails, the error is logged but not propagated
    (error isolation).
    """
    try:
        event = parse_event(method, params)
    except Exception as e:
        logger.warning(
            "Failed to parse event %s: %s: %s — passing raw params",
            method,
            type(e).__name__,
            e,
        )
        event = params  # type: ignore[assignment]
    handlers = list(self._handlers.get(method, []))

    for handler in handlers:
        try:
            await handler(event)
        except Exception as e:
            logger.error(
                "Handler error for %s: %s: %s",
                method,
                type(e).__name__,
                e,
                exc_info=True,
            )

AsyncHandler module-attribute 

AsyncHandler = Callable[[Any], Awaitable[None]]

Type for an async handler that receives an event.

Subscription dataclass

Handle to unsubscribe from an event.

Source code in bidiwave/events/handlers.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@dataclass
class Subscription:
    """Handle to unsubscribe from an event."""

    event_type: str
    handler: AsyncHandler
    _dispatcher_ref: weakref.ReferenceType[EventDispatcher] = field(
        repr=False, compare=False
    )

    def __init__(
        self, event_type: str, handler: AsyncHandler, dispatcher: EventDispatcher
    ) -> None:
        self.event_type = event_type
        self.handler = handler
        self._dispatcher_ref = weakref.ref(dispatcher)

    def unsubscribe(self) -> None:
        """Unsubscribes this handler from its originating dispatcher.

        No-op if the dispatcher has already been garbage-collected.
        """
        dispatcher = self._dispatcher_ref()
        if dispatcher is not None:
            dispatcher.off(self)

unsubscribe 

unsubscribe() -> None

Unsubscribes this handler from its originating dispatcher.

No-op if the dispatcher has already been garbage-collected.

Source code in bidiwave/events/handlers.py 
34
35
36
37
38
39
40
41
def unsubscribe(self) -> None:
    """Unsubscribes this handler from its originating dispatcher.

    No-op if the dispatcher has already been garbage-collected.
    """
    dispatcher = self._dispatcher_ref()
    if dispatcher is not None:
        dispatcher.off(self)