or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

index.md
tile.json

tessl/pypi-zope-event

Very basic event publishing system with synchronous dispatching for Python applications

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/zope.event@5.1.x

To install, run

npx @tessl/cli install tessl/pypi-zope-event@5.1.0

index.mddocs/

zope.event

A 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.

Package Information

  • Package Name: zope.event
  • Language: Python
  • Installation: pip install zope.event
  • Python Versions: 3.9+

Core Imports

import zope.event

For direct access to core functions:

from zope.event import notify, subscribers

For class-based event handling:

from zope.event.classhandler import handler

Basic Usage

import 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 in

Capabilities

Event Publishing

Core 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
    """

Subscriber Management

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)
    """

Class-Based Event Handling

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]]]
    """

Types

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]

Error Handling

  • Subscriber exceptions: Any exception raised by a subscriber during notify() is propagated immediately without calling remaining subscribers
  • Handler registration: The class-based handler system automatically registers its dispatcher with the global subscribers list on first use
  • Event inheritance: Class-based handlers are called in method resolution order, supporting proper inheritance patterns

Advanced Usage

Multiple Event Types

import 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: LoginEvent

Custom Subscriber Management

import 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