Very basic event publishing system with synchronous dispatching for Python applications
npx @tessl/cli install tessl/pypi-zope-event@5.1.0A minimalist event system providing simple synchronous event publishing for Python applications. The package enables publish-subscribe patterns where applications can notify subscribers of events without requiring knowledge of who is listening, serving as a foundation for building more sophisticated event dispatching systems.
pip install zope.eventimport zope.eventFor direct access to core functions:
from zope.event import notify, subscribersFor class-based event handling:
from zope.event.classhandler import handlerimport zope.event
# Define a simple event handler
def my_handler(event):
print(f"Received event: {event}")
# Register the handler
zope.event.subscribers.append(my_handler)
# Create and publish an event
event_data = {"type": "user_login", "user_id": 123}
zope.event.notify(event_data)
# Output: Received event: {'type': 'user_login', 'user_id': 123}
# Class-based event handling
from zope.event.classhandler import handler
class UserLoginEvent:
def __init__(self, user_id):
self.user_id = user_id
# Register handler for specific event class
@handler(UserLoginEvent)
def handle_login(event):
print(f"User {event.user_id} logged in")
# Publish class-based event
login_event = UserLoginEvent(456)
zope.event.notify(login_event)
# Output: User 456 logged inCore event publishing functionality that allows applications to notify all registered subscribers of events synchronously.
def notify(event):
"""
Notify all subscribers of the given event.
Args:
event: Any object to be published as an event
Returns:
None
Raises:
Any exception raised by subscribers is propagated without
running remaining subscribers
"""Global registry for event subscribers that any callable can be added to for receiving event notifications.
subscribers: list
"""
Global list of event subscribers.
Applications append callable subscribers to this list. Each subscriber
must accept a single argument (the event object). Subscribers are called
in the order they appear in the list.
Example:
def my_handler(event):
print(f"Got event: {event}")
zope.event.subscribers.append(my_handler)
"""Enhanced event handling system that registers handlers for specific event classes, supporting inheritance and method resolution order dispatching.
def handler(event_class, handler_=None, _decorator=False):
"""
Register an event handler for a specific event class.
Can be used as a decorator or called directly with class and handler.
Handlers are called based on the event object's class hierarchy using
method resolution order.
Args:
event_class: type - the event class to handle
handler_: callable, optional - the handler function
_decorator: bool, internal - indicates decorator usage
Returns:
callable: decorator function when handler_ is None
callable: the handler function when _decorator is True
None: when called directly with both event_class and handler_
Examples:
# Direct registration
handler(MyEvent, my_handler)
# Decorator usage
@handler(MyEvent)
def my_handler(event):
pass
"""
def dispatch(event):
"""
Internal dispatcher function that calls handlers for event classes.
This function is automatically registered as a subscriber when the first
handler is registered. It iterates through the event's class hierarchy
(method resolution order) and calls all registered handlers for each class.
Args:
event: Any - the event object to dispatch
Returns:
None
"""
# Global registry of event class handlers
registry: dict
"""
Internal registry mapping event classes to their handler lists.
This dictionary maps event classes (types) to lists of handler functions.
Handlers are stored in registration order and called in that order for
each class in the event's method resolution order.
Type: Dict[type, List[Callable[[Any], None]]]
"""from typing import Any, Callable, Dict, List, Optional, Union
# Event objects can be any Python object
Event = Any
# Subscriber functions must match this signature
Subscriber = Callable[[Any], None]
# Event class for class-based handling
EventClass = type
# Handler function signature
Handler = Callable[[Any], None]
# Registry type for internal use
Registry = Dict[type, List[Handler]]
# Handler decorator return type
HandlerDecorator = Callable[[Handler], Handler]notify() is propagated immediately without calling remaining subscribersimport zope.event
from zope.event.classhandler import handler
# Define event hierarchy
class BaseEvent:
pass
class UserEvent(BaseEvent):
def __init__(self, user_id):
self.user_id = user_id
class LoginEvent(UserEvent):
pass
class LogoutEvent(UserEvent):
pass
# Register handlers at different levels
@handler(BaseEvent)
def log_all_events(event):
print(f"Event: {event.__class__.__name__}")
@handler(UserEvent)
def track_user_activity(event):
print(f"User activity: {event.user_id}")
@handler(LoginEvent)
def handle_login(event):
print(f"Login: {event.user_id}")
# Publishing events triggers appropriate handlers
zope.event.notify(LoginEvent(123))
# Output (handlers called in MRO order, then registration order):
# Login: 123
# User activity: 123
# Event: LoginEventimport zope.event
# Save and restore subscriber state
original_subscribers = zope.event.subscribers[:]
zope.event.subscribers.clear()
# Add temporary subscribers
def temp_handler(event):
print("Temporary handler")
zope.event.subscribers.append(temp_handler)
# Use the system
zope.event.notify("test event")
# Restore original state
zope.event.subscribers[:] = original_subscribers