Skip to main content

No project description provided

Project description



A simple and efficient implementation of the observable pattern.


What sets it apart is that it doesn't represents events as a combination of a name and arguments, but instead operates on classes.

Instead of using names like "on_connect" the library encourages the use of a "ConnectEvent" class which has the arguments as its attributes. Instead of listening to a meaningless name observers instead use the event type (the class). When emitting an event we then use instances of the event class.

Apart from other benefits this especially helps with typings and eliminates the issue of having to know the function signature for each event, as the only argument is the event instance.

Using the built-in dataclasses makes it easy to avoid writing boiler-plate code for each event.


import asyncio
import dataclasses

import aiobservable

class ConnectEvent:
    user_id: int
    user_name: str

async def main():
    observable = aiobservable.Observable()

    def on_connect(event: ConnectEvent) -> None:
        print(f"{event.user_name} connected!")

    observable.on(ConnectEvent, on_connect)

    event = ConnectEvent(1, "Simon")

    # emit returns a future which resolves to None when all observers
    # are done handling the event
    await observable.emit(event)


You can install the library from PyPI:

pip install aiobservable

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aiobservable-0.2.2.tar.gz (9.0 kB view hashes)

Uploaded source

Built Distribution

aiobservable-0.2.2-py3-none-any.whl (10.4 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Fastly Fastly CDN Google Google Object Storage and Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page