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).
Event and Callback descriptors do not work when
__slots__ are defined.
If slots are defined a
InstanceHasNoDictError will be raised on access
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
Listener lists (providing their event signatures match) can be appended to another event.