EventsΒΆ

A simple framework to publish events or callbacks to subscribed listener(s).

Async events/callbacks are also supported via the AsyncEvent and AsyncCallback descriptors.

An event can have multiple listening functions where as callbacks can only have a single function bound (assigning a second will remove the previous binding).

Note

Event and Callback descriptors do not work when __slots__ are defined. If slots are defined a InstanceHasNoDictError will be raised on access

Example:

class MyClass:
    started = Event[Callable[[], None]]()
    new_message = AsyncCallback[Callable[[str], Awaitable]]()

    def start(self):
        self.started()

    async def process_message(self, message):
        await self.handle_message(message)


instance = MyClass()

@listen_to(instance.started)
def on_started():
    pass

@bind_to(instance.new_message)
async def on_new_message(message: str):
    pass

Hint

Listener lists (providing their event signatures match) can be appended to another event.