Observer pattern made muy facil
Project description
pyevento
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)
Built Distribution
evento-2.0.1-py3-none-any.whl
(7.2 kB
view hashes)