event

Version: 8.0.8

wexample-event provides two mix-ins — EventDispatcherMixin and EventListenerMixin — that add priority-ordered, thread-safe event dispatch to any Python class without a dedicated event bus. Listeners register by name with add_event_listener or with the @EventListenerMixin.on decorator, and events are dispatched synchronously (dispatch) or asynchronously (dispatch_async), with optional parent-bubbling when _enable_bubbling is enabled. It is aimed at Python developers who want decoupled, observer-pattern communication in library or application code, installable with pip install wexample-event.

Table of Contents

• Installation
• Quickstart
• Tests
• Architecture
• Integration in the Suite
• Dependencies
• Versioning & Compatibility Policy
• License
• About us
• Known Limitations & Roadmap
• Status & Compatibility
• Useful Links
• Migration Notes

Installation

pip install wexample-event

Requires Python >=3.10.

Quickstart

Install from PyPI:

pip install wexample-event

Mix EventDispatcherMixin into any class, register a listener with add_event_listener, then fire it with dispatch:

from wexample_event.common.dispatcher import EventDispatcherMixin
from wexample_event.dataclass.event import Event

class App(EventDispatcherMixin):
    pass

app = App()

def on_click(event: Event) -> None:
    print(event.name, event.payload)

app.add_event_listener("user.click", on_click)
app.dispatch("user.click", payload={"button": "submit"})
# user.click {'button': 'submit'}

dispatch returns the Event that was sent. The callback receives that same object; event.name is the string you dispatched and event.payload is the mapping you passed in.

A runnable set of examples covering multiple listeners, priorities, decorators, and once-only listeners is in examples/various/simple_event_example.py.

Tests

This project uses pytest for testing and pytest-cov for code coverage analysis.

Installation

First, install the required testing dependencies:

.venv/bin/python -m pip install pytest pytest-cov

Basic Usage

Run all tests with coverage:

.venv/bin/python -m pytest --cov --cov-report=html

Common Commands

# Run tests with coverage for a specific module
.venv/bin/python -m pytest --cov=your_module

# Show which lines are not covered
.venv/bin/python -m pytest --cov=your_module --cov-report=term-missing

# Generate an HTML coverage report
.venv/bin/python -m pytest --cov=your_module --cov-report=html

# Combine terminal and HTML reports
.venv/bin/python -m pytest --cov=your_module --cov-report=term-missing --cov-report=html

# Run specific test file with coverage
.venv/bin/python -m pytest tests/test_file.py --cov=your_module --cov-report=term-missing

Viewing HTML Reports

After generating an HTML report, open htmlcov/index.html in your browser to view detailed line-by-line coverage information.

Coverage Threshold

To enforce a minimum coverage percentage:

.venv/bin/python -m pytest --cov=your_module --cov-fail-under=80

This will cause the test suite to fail if coverage drops below 80%.

Architecture

The package is split into two sub-packages under src/wexample_event/: dataclass/ holds immutable or near-immutable data containers; common/ holds the mixins and enums that carry behaviour. Nothing in dataclass/ imports from common/; the dependency runs one way.

Dataclasses

src/wexample_event/dataclass/event.py — Event is a frozen=True, slots=True dataclass. Every dispatch produces or passes through exactly one Event instance; its fields (name, payload, metadata, source, timestamp) are set at construction and never mutated. with_update(**changes) and derive(name, **changes) return new instances via dataclasses.replace.

src/wexample_event/dataclass/listener_record.py — ListenerRecord is a mutable slots=True dataclass created once per add_event_listener call. It carries the callable, the once flag, the numeric priority, and a monotonically increasing order integer used to maintain FIFO ordering within the same priority tier. The EventCallback type alias defined here (Callable[[Event], Awaitable[None] | None]) is the contract every listener must satisfy.

src/wexample_event/dataclass/listener_spec.py — ListenerSpec is a frozen=True, slots=True dataclass that records one application of the @on decorator: the event name, priority, and once flag. It lives on the decorated function as a tuple attribute and is read at bind time, never at dispatch time.

Common

src/wexample_event/common/priority.py — EventPriority is an IntEnum with three named tiers (HIGH = 100, NORMAL = 0, LOW = -100). DEFAULT_PRIORITY is EventPriority.NORMAL. Any plain integer is also accepted wherever a priority is expected.

src/wexample_event/common/listener_state.py — ListenerState is a __slots__ class (not a dataclass) that EventListenerMixin stores on each bound instance. It holds two fields: dispatcher (the EventDispatcherMixin currently bound to) and bindings (a list of (event_name, callback) pairs recorded at bind time so they can be unregistered later).

src/wexample_event/common/dispatcher.py — EventDispatcherMixin is the dispatcher. It stores all runtime state lazily in the instance __dict__ under three private keys: a dict[str, list[ListenerRecord]] keyed by event name, a threading.RLock, and an itertools.count sequence. Lazy initialisation means a class that mixes this in pays no cost until the first listener is registered.

src/wexample_event/common/listener.py — EventListenerMixin is the declarative listener side. The @on class method decorates a method by appending a ListenerSpec to the function's __event_listener_specs__ attribute. bind_to_dispatcher walks type(self).__mro__, collects every decorated method, and registers bound callbacks on the dispatcher, storing the pairs in a ListenerState. unbind_from_dispatcher replays those pairs in reverse through remove_event_listener.

Call path through a synchronous dispatch

1. dispatch(event_or_str, *, payload, metadata, source) on EventDispatcherMixin calls _snapshot_listeners.
2. _snapshot_listeners calls _coerce_event, which either passes an Event instance through unchanged or constructs one from the string name, defaulting source to self. It then acquires the lock, copies the bucket for that event name into a plain list, and releases the lock.
3. Callback loop — the snapshot list is iterated outside the lock. For each ListenerRecord, record.callback(dispatched_event) is called synchronously. If the callback is a coroutine function, a RuntimeError is raised immediately rather than silently producing an unawaited coroutine.
4. Once cleanup — any record whose once flag is True is collected and then removed via remove_event_listener after the loop completes.
5. Bubbling — if _enable_bubbling is True on the class and _get_bubbling_parent() returns a non-None dispatcher, the same Event object is forwarded to parent.dispatch(dispatched_event), repeating the full sequence up the tree until a node returns None from _get_bubbling_parent.

dispatch_async follows the identical path but calls await result when inspect.isawaitable(result) is true, and forwards to parent.dispatch_async during bubbling.

Listener ordering

Within a bucket, ListenerRecord entries are kept sorted by (-priority, order). Higher numeric priority runs first; among equal-priority listeners, registration order (FIFO) is preserved. The sort happens inside add_event_listener under the lock, so the snapshot taken at dispatch time is already in the correct order.

Integration in the Suite

This package is part of the Wexample Suite — a collection of high-quality, modular tools designed to work seamlessly together across multiple languages and environments.

Related Packages

The suite includes packages for configuration management, file handling, prompts, and more. Each package can be used independently or as part of the integrated suite.

Visit the Wexample Suite documentation for the complete package ecosystem.

Dependencies

• wexample-helpers: >=20.2.0

Versioning & Compatibility Policy

Wexample packages follow Semantic Versioning (SemVer):

• MAJOR: Breaking changes
• MINOR: New features, backward compatible
• PATCH: Bug fixes, backward compatible

We maintain backward compatibility within major versions and provide clear migration guides for breaking changes.

License

This project is licensed under the MIT License - see the LICENSE file for details.

Free to use in both personal and commercial projects.

About us

Wexample stands as a cornerstone of the digital ecosystem — a collective of seasoned engineers, researchers, and creators driven by a relentless pursuit of technological excellence. More than a media platform, it has grown into a vibrant community where innovation meets craftsmanship, and where every line of code reflects a commitment to clarity, durability, and shared intelligence.

This packages suite embodies this spirit. Trusted by professionals and enthusiasts alike, it delivers a consistent, high-quality foundation for modern development — open, elegant, and battle-tested. Its reputation is built on years of collaboration, refinement, and rigorous attention to detail, making it a natural choice for those who demand both robustness and beauty in their tools.

Wexample cultivates a culture of mastery. Each package, each contribution carries the mark of a community that values precision, ethics, and innovation — a community proud to shape the future of digital craftsmanship.

Known Limitations & Roadmap

Current limitations and planned features are tracked in the GitHub issues.

See the project roadmap for upcoming features and improvements.

Status & Compatibility

Maturity: Production-ready

Python Support: >=3.10

OS Support: Linux, macOS, Windows

Status: Actively maintained

Useful Links

• Homepage: https://github.com/wexample/python-event
• Documentation: docs.wexample.com
• Issue Tracker: https://github.com/wexample/python-event/issues
• Discussions: https://github.com/wexample/python-event/discussions
• PyPI: pypi.org/project/wexample-event

Migration Notes

When upgrading between major versions, refer to the migration guides in the documentation.

Breaking changes are clearly documented with upgrade paths and examples.