events-manager 0.2.2

Python implementation for the Events Management system (aka Observer pattern)

Events Manager

Python implementation of the Events Management system (aka Observer pattern).

Getting started

Requirements

• Python >= 3.8

Installation

pip install events-manager

Usage

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

./test

Style Check

./lint

License

Distributed under the MIT License. See LICENSE file for more information.