pyee: Python Event Emitter Library

raw JSON →
13.0.1 verified Tue May 12 auth: no python install: verified quickstart: verified

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.

pip install pyee
error ModuleNotFoundError: No module named 'pyee'
cause The `pyee` library is not installed in the Python environment, or the environment where it's installed is not the one being used.
fix
Install the library using pip: pip install pyee
error ImportError: cannot import name 'EventEmitter' from 'pyee'
cause In newer versions of `pyee` (post-8.x, specifically since version 9.0.0), the core `EventEmitter` class and its specialized asynchronous/threaded variants are located in submodules like `pyee.base`, `pyee.asyncio`, `pyee.twisted`, etc., rather than directly under the top-level `pyee` package.
fix
Import EventEmitter from pyee.base for the basic emitter, or from specific submodules like pyee.asyncio for an asyncio-compatible emitter: from pyee.base import EventEmitter (for synchronous) or from pyee.asyncio import AsyncIOEventEmitter (for asyncio).
error DeprecationWarning: pyee.EventEmitter is deprecated and will be removed in a future major version; you should instead use either pyee.AsyncIOEventEmitter, pyee.TwistedEventEmitter, pyee.ExecutorEventEmitter, pyee.TrioEventEmitter, or pyee.BaseEventEmitter.
cause The direct import and usage of `pyee.EventEmitter` as the primary event emitter class is deprecated. Users are encouraged to use the more specific and often more robust emitter types available in dedicated submodules, especially when working with asynchronous or threaded contexts.
fix
Replace from pyee import EventEmitter with from pyee.base import EventEmitter for a basic synchronous emitter, or use a specialized emitter like from pyee.asyncio import AsyncIOEventEmitter if working with asyncio.
error TypeError: constraints must be a tuple
cause This error occurs when using `pyee` with Python 3.12 due to a change in how `TypeVar` (a feature used internally by `pyee` for type hinting) handles its 'constraints' argument in Python 3.12. Older `pyee` versions passed `constraints` as a single type, but Python 3.12 requires it to be a tuple.
fix
Upgrade pyee to a version that officially supports Python 3.12. For example, pyee version 12.0.0 and above addresses this compatibility issue. Use pip install --upgrade pyee.
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.
python os / libc status wheel install import disk
3.10 alpine (musl) wheel - 0.01s 18.2M
3.10 alpine (musl) - - 0.01s 18.2M
3.10 slim (glibc) wheel 1.6s 0.01s 19M
3.10 slim (glibc) - - 0.01s 19M
3.11 alpine (musl) wheel - 0.02s 20.1M
3.11 alpine (musl) - - 0.02s 20.1M
3.11 slim (glibc) wheel 1.7s 0.02s 21M
3.11 slim (glibc) - - 0.02s 21M
3.12 alpine (musl) wheel - 0.02s 12.0M
3.12 alpine (musl) - - 0.02s 12.0M
3.12 slim (glibc) wheel 1.5s 0.02s 12M
3.12 slim (glibc) - - 0.02s 12M
3.13 alpine (musl) wheel - 0.01s 11.7M
3.13 alpine (musl) - - 0.02s 11.6M
3.13 slim (glibc) wheel 1.5s 0.01s 12M
3.13 slim (glibc) - - 0.01s 12M
3.9 alpine (musl) wheel - 0.01s 17.7M
3.9 alpine (musl) - - 0.01s 17.7M
3.9 slim (glibc) wheel 1.9s 0.01s 18M
3.9 slim (glibc) - - 0.01s 18M

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