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 fromsaul_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
- Define your custom events by inheriting from
Event. - Create an
EventPublisherinstance. - Use the
event_handlerdecorator orsubscribemethod to register handlers. - 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
EventPublisheris 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
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
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
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
4ca7755f079e9b4df25abaaa2199be0d98131a090196b691367878219ab0fb66
|
|
| MD5 |
e0dfb6b2750dd4ed8f965612720de97c
|
|
| BLAKE2b-256 |
e264bf23b989fe1419069697f141ef2df1ffaf1b2f5cb4c386e8cbe33efac583
|
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
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
cc0a2848e357e51169c863844af08e77c21da93f92a3ca6eefe6f4c74e758a9c
|
|
| MD5 |
879768f58a3849688c4ab9b328b0c133
|
|
| BLAKE2b-256 |
4f0cadb6edc338cca90f5c0db902bc5c533e47ca8c9772b780ec0be7c777784e
|