Skip to main content

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

Project description

Event Framework

A lightweight event-driven framework in Python that allows you to define events, subscribe handlers, and publish them through a central publisher.
It is designed with clarity, maintainability, and extensibility in mind.


Got it Tony — here’s the imports section rewritten in English Markdown, clear and concise, showing exactly how users should import from your saul_events package:


Imports

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

from py-events import (
    EventPublisher,
    EventMiddleware,
    MiddlewarePipeline,
    event_handler,
    Event
)

Exceptions are imported from the exceptions submodule:

from py-events.exceptions import (
    MissingEventError,
    EventSubclassRequiredError,
    InvalidEventTypeError,
    UnannotatedEventParameterError
)

This makes it clear:

  • Non-error classes/functions (EventPublisher, EventMiddleware, MiddlewarePipeline, event_handler, Event) are imported directly from saul_events.
  • Errors/exceptions are imported from py_event.exceptions.

⚙️ 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:
        @attr.frozen
        class UserLoginEvent(Event):
            username: str

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

    """

Usage Example:

      @attr.frozen
        class UserLoginEvent(Event):
            username: str
                    # Example instantiation
        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()

Got it Tony — here’s the full documentation and examples in English Markdown, ready to copy‑paste directly:


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 event import Event
from 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

✅ Conclusion

This framework provides a simple yet powerful way to implement event-driven programming in Python.
It is extensible, easy to maintain, and supports both explicit and annotation-based handler registration.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

saul_events-0.1.0.tar.gz (4.4 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

saul_events-0.1.0-py3-none-any.whl (3.7 kB view details)

Uploaded Python 3

File details

Details for the file saul_events-0.1.0.tar.gz.

File metadata

  • Download URL: saul_events-0.1.0.tar.gz
  • Upload date:
  • Size: 4.4 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.7

File hashes

Hashes for saul_events-0.1.0.tar.gz
Algorithm Hash digest
SHA256 4ca7755f079e9b4df25abaaa2199be0d98131a090196b691367878219ab0fb66
MD5 e0dfb6b2750dd4ed8f965612720de97c
BLAKE2b-256 e264bf23b989fe1419069697f141ef2df1ffaf1b2f5cb4c386e8cbe33efac583

See more details on using hashes here.

File details

Details for the file saul_events-0.1.0-py3-none-any.whl.

File metadata

  • Download URL: saul_events-0.1.0-py3-none-any.whl
  • Upload date:
  • Size: 3.7 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.7

File hashes

Hashes for saul_events-0.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 cc0a2848e357e51169c863844af08e77c21da93f92a3ca6eefe6f4c74e758a9c
MD5 879768f58a3849688c4ab9b328b0c133
BLAKE2b-256 4f0cadb6edc338cca90f5c0db902bc5c533e47ca8c9772b780ec0be7c777784e

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page