Events Manager
Python implementation of the Events Management system (aka Observer pattern).
Getting started
Requirements
- Python >= 3.8
Installation
pip install events-managerUsage
Listening and emitting an Event
Listener is invoked when the event is emitted.
from asyncio import run, sleep
from events_manager import Event, emit, listen
class FooEvent(Event):
pass
def foo_listener(event: FooEvent):
print(f"'foo_listener' invoked with {event}")
async def main():
listen(FooEvent, foo_listener)
print("Emitting event...")
emit(FooEvent())
# do the other stuff...
await sleep(1)
if __name__ == '__main__':
run(main())Output:
Emitting event...
'foo_listener' invoked with <__main__.FooEvent object at 0x7fdce6f5a0a0>
Process finished with exit code 0
sync and async listeners
from asyncio import run, sleep
from events_manager import Event, emit, listen
class FooEvent(Event):
pass
async def foo_listener(event: FooEvent):
print(f"'foo_listener' invoked with {event}")
async def main():
listen(FooEvent, foo_listener)
print("Emitting event...")
emit(FooEvent())
# do the other stuff...
await sleep(1)
if __name__ == '__main__':
run(main())Output:
Emitting event...
'foo_listener' invoked with <__main__.FooEvent object at 0x7f81e76ad0a0>
Process finished with exit code 0
args and kwargs support
The listen method supports also args and kwargs that will be passed to the listened listener.
from asyncio import run, sleep
from events_manager import Event, emit, listen
class FooEvent(Event):
pass
async def foo_listener(event: FooEvent, bar, baz):
print(f"'foo_listener' invoked with {event}, {bar} and {baz}")
async def main():
listen(FooEvent, foo_listener, False, 'bar', baz='baz')
print("Emitting event...")
emit(FooEvent())
# do the other stuff...
await sleep(1)
if __name__ == '__main__':
run(main())Output:
Emitting event...
'foo_listener' invoked with <__main__.FooEvent object at 0x7fdbd0fa50d0>, bar and baz
Process finished with exit code 0
Register a listener with @on decorator
Instead of calling listen method, you can also use the @on decorator.
from asyncio import run, sleep
from events_manager import Event, emit, on
class FooEvent(Event):
pass
@on(FooEvent)
def foo_listener(event: FooEvent):
print(f"'foo_listener' invoked with {event}")
async def main():
print("Emitting event...")
emit(FooEvent())
# do the other stuff...
await sleep(1)
if __name__ == '__main__':
run(main())Output:
Emitting event...
'foo_listener' invoked with <__main__.FooEvent object at 0x7fa0a9a47100>
Process finished with exit code 0
Unregister a listener
Call unregister method passing the event type that you want to stop listening and the listener.
from asyncio import run, sleep
from events_manager import Event, emit, listen, unregister
class FooEvent(Event):
pass
def foo_listener(event: FooEvent):
print(f"'foo_listener' invoked with {event}")
async def main():
listen(FooEvent, foo_listener)
print("Emitting first event...")
emit(FooEvent())
# let the event be processed
await sleep(1)
unregister(FooEvent, foo_listener)
print("Emitting second event...")
emit(FooEvent())
# do the other stuff...
await sleep(1)
if __name__ == '__main__':
run(main())Output:
Emitting first event...
'foo_listener' invoked with <__main__.FooEvent object at 0x7f92c79b9070>
Emitting second event...
Process finished with exit code 0
Development
Run Tests
./testStyle Check
./lintLicense
Distributed under the MIT License. See LICENSE file for more information.
