Eventlib-py

Eventlib-py is a small & simple event framework for Python that can be used to decouple your code.

Feature overview:

• Fast - Pre-computed event chains for fast event emission.
• Lightweight - No extra dependencies.
• Asynchronous - Full support for asynchronous event handlers and context managers.
• Priority Ordering - Control the order in which event handlers are called.
• Monitoring - Use context managers to monitor event processing.
• Compatible - Works with almost anything, use: dataclasses, Pydantic or attrs.

import asyncio
import dataclasses

import eventlib


@dataclasses.dataclass
class MyEvent(eventlib.BaseEvent):
    value: str


@eventlib.subscribe()
async def on_my_event(event: MyEvent):
    print(f"Received: {event.value}")


asyncio.run(MyEvent("Hello, world!").emit_async())  # Prints: "Received: Hello, world!"

Usage

Event Inheritance

import eventlib


class MyEvent(eventlib.BaseEvent):
    pass


class SubEvent(MyEvent):
    pass


@eventlib.subscribe()
def on_base_event(event: MyEvent):
    print("Received event", event.__class__.__name__)


SubEvent().emit()  # Prints: "Received event SubEvent"

Priority Ordering

import eventlib


class MyEvent(eventlib.BaseEvent):
    pass


@eventlib.subscribe(priority=-1)
def first():
    print("first")


@eventlib.subscribe()  # default: priority = 0
def second():
    print("second")


@eventlib.subscribe(priority=1)
def third():
    print("third")


MyEvent().emit()  # Prints: "first", "second", "third"

Context Managers

import contextlib
import eventlib


class MyEvent(eventlib.BaseEvent):
    pass


@eventlib.subscribe(priority=-1000)  # Ensure that this is called first
@contextlib.contextmanager
def monitor(event: MyEvent):
    print("Event received")
    try:
        yield
    finally:
        print("Event processed")


@eventlib.subscribe()
def on_event(event: MyEvent):
    print("on_event")


MyEvent().emit()  # Prints: "Event received", "on_event", "Event processed"

Asyncio

import asyncio
import contextlib
import eventlib


class MyEvent(eventlib.BaseEvent):
    pass


@eventlib.subscribe(priority=-1000)  # Ensure that this is called first
@contextlib.asynccontextmanager
async def monitor(event: MyEvent):
    print("Event received")
    try:
        yield
    finally:
        print("Event processed")


@eventlib.subscribe()
async def async_on_event(event: MyEvent):
    print("async_on_event")


@eventlib.subscribe()
def on_event(event: MyEvent):
    print("on_event")


asyncio.run(MyEvent().emit_async())  # Prints: "Event received", "async_on_event", "on_event", "Event processed"

Benchmarks

The benchmark directory contains code to measure the performance of the eventlib-py library and compare it with a hard-coded reference implementation in Python.

Benchmark case_all

The following table shows the overhead of the eventlib-py library in the case_all benchmark. It's a mixed benchmark with all kinds of sync & async event handlers and context managers.

The overhead per call is the additional time that is needed to call the event handlers introduced by the eventlib-py library. The setup time is the additional nonrecurring overhead for subscribing the event handlers in the event system. It shows that in the worst case a 15% overhead per call is introduced. The expected median overhead is around 7% versus hard-coded event handling.

Development

Use poetry to setup the development environment.

poetry install --with=dev
poetry shell

Run the auto-formatter, checks and linter:

black .
isort .
mypy .
pylint .

Run the tests with coverage:

pytest --cov

Contributing

Contributions are welcome.

Please follow the commit convention https://www.conventionalcommits.org/en/v1.0.0/.

License

Dual-licensed under the terms of either the Apache License 2.0 or the MIT license.

SPDX-License-Identifier: (Apache-2.0 OR MIT)