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 whereas 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.