API Reference

class pyesys.event.Event(*, allow_duplicates: bool = True, error_handler: ErrorHandler | None = None)

A thread-safe event dispatcher with comprehensive features:

• Weak-reference support for bound methods (automatic cleanup)

• Runtime signature checking via example function

• Configurable error handling with consistent behavior

• Introspection: handler_count, handlers list

• Duplicate subscription control

• Mixed sync/async support with proper resource management

• Performance optimizations with lazy cleanup

Generic P: parameter specification for handler arguments.

class Listener(outer: Event)

Provides a restricted interface to subscribe/unsubscribe handlers to the outer Event without exposing emission or clearing capabilities.

This separation ensures subscribers cannot accidentally trigger events or clear all handlers.

handler_count() → int

Number of currently alive handlers.

Returns:

Count of active handlers.

subscribe(handler: Callable[[P], None] | Iterable[Callable[[P], None]]) → None

Alternative to += operator for subscribing handlers.

Supports single callables or iterables (list, tuple, set).

Parameters:

handler – Callable or iterable of callables to subscribe.

Raises:

TypeError – If handler is not callable.

unsubscribe(handler: Callable[[P], None] | Iterable[Callable[[P], None]]) → None

Alternative to -= operator for unsubscribing handlers.

Supports single callables or iterables (list, tuple, set).

Parameters:

handler – Callable or iterable of callables to unsubscribe.

Raises:

TypeError – If handler is not callable.

clear() → None

Remove all subscribed handlers from this Event.

This is useful for cleanup or resetting event state.

emit(*args: ~typing.~P, **kwargs: ~typing.~P) → None

Emit the event synchronously, invoking all subscribed handlers.

Error handling isolates exceptions: one handler’s exception does not prevent others from running. Dead bound-method handlers are automatically removed during cleanup.

Async handlers that return coroutines are silently ignored in sync emission.

Parameters:

• args – Positional arguments matching signature P.

• kwargs – Keyword arguments matching signature P.

async emit_async(*args: ~typing.~P, **kwargs: ~typing.~P) → None

Asynchronously emit the event, invoking all subscribed handlers concurrently.

Handlers that are coroutine functions will be awaited; regular callables will be scheduled on the default executor. All exceptions are consistently routed to the configured error_handler.

Parameters:

• args – Positional arguments matching signature P.

• kwargs – Keyword arguments matching signature P.

handler_count() → int

Number of currently alive handlers subscribed.

This property triggers cleanup of dead handlers for accurate counting.

Returns:

Count of active handlers.

property handlers: List[Callable[[P], None]]

Return a list of currently alive handler callables.

This property triggers cleanup and reconstructs callable references.

Returns:

List of active handler functions or bound methods.

classmethod new(example: Callable[[P], None], *, allow_duplicates: bool = True, error_handler: ErrorHandler | None = None) → Tuple[Event, Listener]

Factory method to create an Event with runtime signature checking.

Parameters:

• example – Example function whose signature defines allowed handler signature.

• allow_duplicates – If True, same handler can be subscribed multiple times.

• error_handler – Custom error handler for exceptions during emission.

Returns:

Tuple of (Event instance, Listener interface).

Raises:

TypeError – If example is not callable.

subscribe_one(handler: Callable[[P], None]) → None

Subscribe a single handler to this Event with validation and duplicate control.

Performs runtime signature checking.

Parameters:

handler – Callable to subscribe.

Raises:

TypeError – If handler is invalid or incompatible.

unsubscribe_one(handler: Callable[[P], None]) → None

Unsubscribe a single handler from this Event.

Parameters:

handler – Callable previously subscribed.

Raises:

TypeError – If handler is not callable.

pyesys.event.create_event(example: Callable[[P], None], *, allow_duplicates: bool = True, error_handler: ErrorHandler | None = None) → Tuple[Event, Listener]

Factory method to create an Event with runtime signature checking.

Parameters:

• example – Example function whose signature defines allowed handler signature.

• allow_duplicates – If True, same handler can be subscribed multiple times.

• error_handler – Custom error handler for exceptions during emission.

Returns:

Tuple of (Event instance, Listener interface).

Raises:

TypeError – If example is not callable.

class pyesys.prop.EventDescriptor(func: Callable[[P], None])

Unified descriptor for both module-level and class-level events.

• If used at module level (no __set_name__), it holds a single global Event.

• If used inside a class (after __set_name__), it manages a per-instance Event.

Features: - Automatic signature detection and validation - Per-instance events for class usage - Global events for module usage - Emitter decorator pattern - Mixed sync/async handler support with proper resource management

emitter(fn: Callable[[P], None]) → Callable[[P], None]

Decorator for the emitter function. Wraps fn so that after running its body, it fires the appropriate Event.

Parameters:

fn – Function to wrap as an emitter.

Returns:

Wrapped function that emits events after execution.

Raises:

TypeError – If fn is not callable.

pyesys.prop.event(func: Callable[[P], None]) → EventDescriptor

Decorator to create a module-level or class-level event.

This unified decorator automatically handles both use cases: - When used at module level: creates a global event - When used in a class: creates per-instance events

Usage: .. code-block:: python

@event def on_something(…):

‘’’Event signature definition’’’ pass

@on_something.emitter def do_action(…):

‘’’Action that triggers the event’’’ # Your logic here pass # Event automatically emitted after this

In a class, this becomes a per-instance event; at module level, it’s a single global event.

Parameters:

func – Function defining the event signature.

Returns:

EventDescriptor that manages the event.

Raises:

TypeError – If func is not callable.

class pyesys.handler.ErrorHandler(*args, **kwargs)

Protocol for custom error handlers used in event emission.

Error handlers receive exceptions that occur during handler execution and can implement custom logging, reporting, or recovery logic.

class pyesys.handler.EventHandler(func: Callable[[P], None])

Wraps a callable handler with advanced features:

• Weak-reference support for bound methods (prevents memory leaks)

• Automatic cleanup when instances are garbage-collected

• Proper equality and hashing for duplicate detection

• Thread-safe callback reconstruction

• Handles both sync and async callables appropriately

This class handles the complexity of managing bound method references while providing a simple callable interface.

get_callback() → Callable[[P], None] | None

Return the current callable, reconstructing bound methods if necessary.

This method handles the complexity of recreating bound methods from their components while checking for instance liveness.

Returns:

The original function or bound method, or None if dead.

get_info() → dict[str, Any]

Get detailed information about this handler for debugging purposes.

Returns:

Dictionary containing handler information.

is_alive() → bool

Check if this handler is still alive and can be invoked.

Returns:

False if it was a bound method and the instance is gone, True otherwise.

is_async() → bool

Check if this handler is async (coroutine function).

Returns:

True if the handler is async.

pyesys.handler.default_error_handler(exception: Exception, handler: Callable[[...], None] | None) → None

Default error handler that prints exceptions to stderr.

This provides basic error reporting without causing the application to crash when event handlers raise exceptions.

Parameters:

• exception – The exception that was raised.

• handler – The handler that raised the exception (may be None).