yapee 1.0.2

Yet Another Python Event Emitter

Yet Another Python Event Emitter

ASAP (as simple as possible), fast, and understandable brokerless, zero-dependent event emitter for Python.

Features

• Supports both synchronous and asynchronous listeners.
• Provides event waiting with predicates and optional timeouts.
• Allows dynamic listener registration and removal.
• Built-in error propagation and handling.
• Enables lazy event emission.

Installation

Project published on PyPi under yapee name.

poetry add yapee

pip install yapee

uv add yapee

Usage

Basic Example

import asyncio

from yapee import EventEmitter

ee = EventEmitter()


@ee.enlist("my_event")
def my_listener(data):
    print(f"Received data: {data}")


ee.emit("my_event", "Hello, EventEmitter!")

Asynchronous Listeners

@ee.enlist("async_event")
async def async_listener(data):
    await asyncio.sleep(1)  # Do some async work
    print(f"Async received: {data}")


async def main():
    ee.emit("async_event", "Hello Async!")
    await asyncio.sleep(2)  # Give time for async execution


asyncio.run(main())

Waiting for an Event

async def trigger_event():
    await asyncio.sleep(2)
    ee.emit("wait_event", "Waited data")


asyncio.create_task(trigger_event())
data = await ee.wait_for("wait_event", timeout=5)
print(f"Waited and got: {data}")

Waiting for an Event with Predicate Function

Pay attention that predicate function must be light and do not block execution.

async def trigger_event():
    await asyncio.sleep(2)
    ee.emit("filtered_event", 49)  # Emit event with data, will not pass predicate, as 49 < 50
    await asyncio.sleep(2)
    ee.emit("filtered_event", 51)  # Will pass predicate and wait_for will return 51


asyncio.create_task(trigger_event())

data = await ee.wait_for("filtered_event", predicate=lambda x: x > 50)
print(f"Received matching event data: {data}")

Removing Listeners

# Removing a specific listener
ee.delist("my_event", my_listener)
ee.delist("my_event", async_listener)

# Removing all listeners for an event
ee.delist_all("my_event")

# Removing all listeners for all events
ee.delist_all()

Error Handling in Listeners

By default, any exception inside a listener will be raised. You can override _on_listener_error to customize error handling:

class MyCustomEmitter(EventEmitter):
    async def _on_listener_error(self, event, listener, args, e):
        print(f"Error in listener {listener} for event {event}: {e}")


my_custom_ee = MyCustomEmitter()

Is there anybody interested in this event?

Sometimes emitting an event is cheap, but preparing its payload is not. Use these helpers to check whether an event has any listeners or waiters before doing expensive work:

ee.has_listeners(event)  # at least one listener is registered for event
ee.has_waiters(event)  # at least one active waiter is waiting for event
ee.has_any(event)  # at least one listener or waiter is here awaiting for event

# So
if ee.has_any("my_event"):
    payload = expensive_work()
    ee.emit("my_event", payload)