pyee: Python Event Emitter Library

13.0.1 · active · verified Sat Mar 28

pyee is a Python library that provides an EventEmitter implementation, heavily inspired by Node.js's EventEmitter. It supports synchronous, asynchronous (asyncio, Twisted, Trio), and threaded event handling. The library is currently in version 13.0.1 and maintains a regular release cadence with major updates roughly annually and patch releases in between.

Warnings

breaking Direct top-level imports of specific EventEmitter subclasses (e.g., `pyee.AsyncIOEventEmitter`, `pyee.TwistedEventEmitter`, `pyee.ExecutorEventEmitter`, `pyee.TrioEventEmitter`) were removed in version 12.0.0.
Fix: Import the specific emitters from their respective submodules. For example, change `from pyee import AsyncIOEventEmitter` to `from pyee.asyncio import AsyncIOEventEmitter`.
deprecated The `PyeeException` class was deprecated in version 12.0.0.
Fix: Use `PyeeError` instead of `PyeeException`. `PyeeError` inherits from `PyeeException`.
breaking Type stub for `twisted.python.Failure` was removed in version 13.0.0, primarily to address typing issues in unsupported Twisted versions. This might affect projects relying on specific type checking behavior.
Fix: Review type checking configurations and potentially update Twisted versions or adjust type ignore rules if encountering new type errors related to `twisted.python.Failure`.
gotcha Error handling differs between `EventEmitter` and its asynchronous counterparts. The base `EventEmitter` does not automatically catch or emit exceptions from its handlers; they must be handled explicitly or an `error` listener provided. `AsyncIOEventEmitter` (and other async emitters) *do* automatically emit exceptions on the 'error' event.
Fix: Always attach an `error` listener (`@ee.on('error')`) to `EventEmitter` instances if unhandled exceptions are a concern, especially for synchronous handlers. For `AsyncIOEventEmitter`, be prepared to handle `error` events for exceptions originating in async handlers.

Install

pip install pyee Install stable version

Imports

EventEmitter
```
from pyee import EventEmitter
```
AsyncIOEventEmitter
```
from pyee.asyncio import AsyncIOEventEmitter
```
Top-level imports for specific emitters were removed in v12.0.0. Import from respective submodules (e.g., pyee.asyncio).

Quickstart

This quickstart demonstrates both the synchronous `EventEmitter` and the `AsyncIOEventEmitter` for handling events. It shows how to register listeners using the `@on` decorator and emit events using the `emit` method. For `AsyncIOEventEmitter`, it includes an example of waiting for handlers to complete and basic error handling.

import asyncio
from pyee import EventEmitter
from pyee.asyncio import AsyncIOEventEmitter

# Synchronous EventEmitter
sync_ee = EventEmitter()

@sync_ee.on('sync_event')
def handle_sync_event(data):
    print(f"Synchronous event received: {data}")

sync_ee.emit('sync_event', 'Hello from sync!')

# Asynchronous AsyncIOEventEmitter
async_ee = AsyncIOEventEmitter()

@async_ee.on('async_event')
async def handle_async_event(data):
    print(f"Asynchronous event received: {data}")
    await asyncio.sleep(0.1) # Simulate async work
    print("Async handler finished.")

@async_ee.on('error')
def on_async_error(e):
    print(f"Async error caught: {e}")

async def main():
    print("Emitting async event...")
    async_ee.emit('async_event', 'Hello from async!')
    # Wait for all handlers to complete (new in v12.1.0)
    await async_ee.wait_for_complete()
    print("All async handlers completed.")

    print("Emitting an error (async)...")
    async_ee.emit('error', ValueError("Something went wrong in async!"))

if __name__ == '__main__':
    asyncio.run(main())

view raw JSON →