py-events-bus 0.1.0

A lightweight, event-driven framework for Python. Provides an EventPublisher, middleware pipeline, and decorator-based event handlers.

---

Imports

The root package is py_event_bus.
From there you can import the main components directly:

from py_event_bus import (
    EventPublisher,
    EventMiddleware,
    MiddlewarePipeline,
    event_handler,
    Event
)

Exceptions are imported from the exceptions submodule:

from py_event_bus.exceptions import (
    MissingEventError,
    EventSubclassRequiredError,
    InvalidEventTypeError,
    UnannotatedEventParameterError
)

---

⚙️ Components

1. Event (event.py)

class Event:
    """
    Immutable base class for all events in the framework.

    Every custom event must inherit from this class. Because the base class
    is decorated with `@attr.frozen`, all subclasses are automatically frozen
    as well. This ensures that event instances cannot be modified after
    creation, preserving consistency and traceability across the system.

    Attributes:
        id (uuid.UUID): A unique identifier automatically generated when
            the event is created.
        timestamp (datetime.datetime): The exact date and time when the
            event instance was created.
        type (str): The name of the event class, useful for distinguishing
            event types during dispatching and logging.
    """

Usage Example:

@attr.frozen
class UserLoginEvent(Event):
    username: str

login_event = UserLoginEvent("John Doe")
print(login_event.id)        # Unique UUID
print(login_event.timestamp) # Creation time
print(login_event.type)      # "UserLoginEvent"

---

2. EventPublisher (publisher.py)

class EventPublisher(Singleton):
    """
    Central event dispatcher and subscription manager.

    The EventPublisher manages event listeners and dispatches events to their
    corresponding handlers. It ensures a single publisher instance across the
    application.

    Methods:
        publish(event: Event): Dispatches an event to all subscribed listeners.
        subscribe(event_class: Type[Event], event_listener: EventListener):
            Subscribes a listener to an event class.
        unsubscribe(event: Type[Event], listener_method: Callable):
            Removes a listener method from an event class.
        unsubscribe_all(): Clears all registered listeners.
    """

Usage Example:

publisher = EventPublisher(debug=True, in_order=True)

# Subscribe a handler
publisher.subscribe(UserLoginEvent, EventListener(order=1, method=handle_login))

# Publish an event
publisher.publish(UserLoginEvent(username="Tony"))

# Unsubscribe a handler
publisher.unsubscribe(UserLoginEvent, handle_login)

# Clear all listeners
publisher.unsubscribe_all()

---

event_handler Decorator (handler.py)

def event_handler(
    publisher: EventPublisher,
    event_class: Type[Event] | FrozenSet[Event] = None,
    order: int = None
):
    """
    Decorator used to register a function as an event handler.

    Supports two modes:
    1. Explicit subscription: Provide `event_class` to subscribe directly.
    2. Annotation-based subscription: If `event_class` is not provided,
       parameter annotations of the function are inspected to determine
       which events to subscribe to.

    Exceptions:
    - UnannotatedEventParameterError:
        Raised if the decorated function has parameters without type annotations.
    - EventSubclassRequiredError:
        Raised if the annotated parameter type is not a subclass of `Event`.
    """

---

✅ Correct Usage Example

@event_handler(publisher, UserLoginEvent, order=1)
def handle_login(event: UserLoginEvent):
    print(f"User {event.username} logged in")

# Or using parameter annotations:
@event_handler(publisher)
def handle_login(event: UserLoginEvent):
    print(f"User {event.username} logged in")

---

⚠️ Example Raising UnannotatedEventParameterError

@event_handler(publisher)
def handle_login(event):  # ❌ Missing type annotation
    print(f"User {event.username} logged in")

---

⚠️ Example Raising EventSubclassRequiredError

class NotAnEvent:
    pass

@event_handler(publisher)
def handle_invalid(event: NotAnEvent):  # ❌ Not a subclass of Event
    print("This should never be executed")

---

Example: Using EventMiddleware

from py_event_bus.event import Event
from py_event_bus.middlewares import EventMiddleware, MiddlewarePipeline

class UserLoginEvent(Event):
    def __init__(self, username: str):
        super().__init__()
        self.username = username

class LoggingMiddleware(EventMiddleware):
    def process(self, event, next_middl):
        print(f"[Start] {event.type}")
        result = next_middl(event)
        print(f"[End] {event.type}")
        return result

pipeline = MiddlewarePipeline()
pipeline.add(LoggingMiddleware(order=1))

---

🚀 Getting Started

1. Define your custom events by inheriting from Event.
2. Create an EventPublisher instance.
3. Use the event_handler decorator or subscribe method to register handlers.
4. Publish events with publisher.publish(event).

---

🧩 Example Workflow

# event.py
@attr.frozen
class UserLoginEvent(Event):
    username: str

# handler.py
@event_handler(publisher, UserLoginEvent, order=1)
def handle_login(event: UserLoginEvent):
    print(f"User {event.username} logged in")

# main.py
publisher.publish(UserLoginEvent(username="Tony"))

---

📖 Notes

• EventPublisher is a Singleton, ensuring only one dispatcher exists.
• Handlers can be executed in order if in_order=True.

class EventListener:
    """
    If you want to manually subscribe events, you must create an instance of this class.

    Example:
        def handle_user_login(event):
            print(f"User {event.username} logged in")

        listener = EventListener(order=1, method=handle_user_login)
        publisher.subscribe(UserLoginEvent, listener)
    """
    def __init__(self, order: int, method: Callable):
        self.method = method
        self.order = order

---