EventsΒΆ

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

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

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

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.