PL-EBS

Peach's Lightweight Event Bus Singleton

Low code, low dependency, simple implementation of an Event Bus

Events

Events are the messages that will be published to topics.

Events included in a events/ package at the root of the project are automatically imported to PL-EBS at start up. It is not a requirement that all Event definitions exist in one package but it is recommended as it makes the code cleaner and easier to manage. It is slighly more efficient as well since the penalty of dynamic class creation and registration are paid at initialization and not during run time.

from event_bus import event_message

# Event that uses the class name as the topic
@event_message
class MyEvent:
    data: str

# Event that uses a specific topic name
@event_message(topic="custom.topic")
class AnotherEvent:
    pass

Publish

Once an event is registered it can be used on the event bus. Events are published by calling the publish(topic: str, **kwargs) method. The key word arguments are used to populate the fields defined in the Event's class.

from event_bus import EventBus

EventBus().publish("MyEvent", data="Important Message")

Handlers

Unlike Events, handlers are not self-registered at initialization. Handlers can be registered by using the @event_handler decorator along with the register_handlers(obj: Any) method or calling EventBus().subscribe(topic: str, handler: Callable[BaseEvent])

Using decorators:

from event_bus import event_handler, register_handlers

class MyClass:

    def __init__(self):
        register_handlers(self)

    @event_handler("MyEvent")
    def handler(self, msg):
        do_something(msg)

Subscribing to a topic directly:

from event_bus import EventBus

def handler(msg):
    do_something(msg)

EventBus().subscribe(topic="MyEvent", handler=handler)