Skip to main content

Observer pattern made muy facil

Project description


Build Status PyPI wheel Github Tag

Python evento package, making the Observer pattern estúpida sencillo.


pip install evento

Basic Usage

from evento import Event

# observers are simply methods
def observer(value: str):
	print("Received: " + value)

# we need to create an instance for every event we want to fire
demo_event: Event[str] = Event()

# subscribe observer to the event
demo_event += observer

# trigger notifications for the event (run all observers)
demo_event('Hello') # => "Received: Hello"
demo_event('World') # => "Received: World"

Typical usage

This is how Events are typically used to decouple code into separate classes;

from evento import Event

class Action:
	def __init__(self, name) -> None: = name
		self.run_event = Event[Action]()

	def run(self) -> None:

class ActionCounter:
	def __init__(self, action: Action) -> None:
		self.count = 0
		# Event.add returns an unsubscribe method
		self._disconnect = action.run_event.add(self._on_action)

	def __del__(self) -> None:
		# observers should make sure to unsubscribe from events when they are done

	def _on_action(self, action: Action) -> None:
		self.count += 1
		print(f"'{}' ran {self.count} times")

action = Action('Foo action')
counter = ActionCounter(action) # => "'Foo action' ran 1 times" # => "'Foo action' ran 2 times"


# setup
event = Event()

def handler(value) -> None:

def setup() -> Callable[[], None]:
	unsubscribe = event.add(handler)
	return unsubscribe

cleanup = setup()

# ... do stuff ...

# cleanup; all the following lines do the same
event -= handler

Async Event

Works the same as Event but takes async subscribers and has to be awaited;

from evento import AsyncEvent
event = AsyncEvent[float]()

async def echo_double(value: float) -> None:
	print(f"double: {value * 2}")

# ...
await event(5.0) # using __call__
await # using fire (same as __call__)
# ...

Signature Event

Since version 2.0.0 evento is typed and Event and SyncEvent are generic classes with a single type; they are 'fired' using a single argument, and all subscribers are expected to take one argument of that type.

To support more complex method signatures, the event and async_event decorator can be used to turn a method into an event;


from evento import event
def multi_arg_event(id: int, message: str, price: float, **opts: Any) -> None:

def observer(id: int, message: str, price: float, **opts: Any) -> None:
    print(f"observer: id={id}, message={message}, price={price}, opts={opts}")

multi_arg_event += observer
multi_arg_event(0, "Hello World!", 9.99, demo=True)


from evento import async_event

async def observer(id: int, message: str, price: float, **opts: Any) -> Any:
    print(f"observer: id={id}, message={message}, price={price}, opts={opts}")

async def multi_arg_event(id: int, message: str, price: float, **opts: Any) -> str:
	return "Done"

# the original method is still invoked after all event subscribers have executed
result = await multi_arg_event(1, "Test-12", 9.99, feature=True)
print(f"Result: {result}") # => Done

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

evento-2.0.1.tar.gz (6.7 kB view hashes)

Uploaded source

Built Distribution

evento-2.0.1-py3-none-any.whl (7.2 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page