Skip to main content

Observer pattern made muy facil

Project description

pyevento

Build Status PyPI wheel Github Tag

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

Install

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:
		self.name = name
		self.run_event = Event[Action]()

	def run(self) -> None:
		self.run_event(self)

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
		self._disconnect()

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

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

Unsubscribe

# setup
event = Event()

def handler(value) -> None:
	print(value)

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

cleanup = setup()

# ... do stuff ...

# cleanup; all the following lines do the same
event -= handler
event.remove(handler)
cleanup()

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}")

event.append(echo_double)
# ...
await event(5.0) # using __call__
await event.fire(10.0) # using fire (same as __call__)
# ...
event.remove(echo_double)

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;

Sync

from evento import event
@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)

Async

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_event
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 Python 3

Supported by

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