Metadata-Version: 2.4
Name: yapee
Version: 1.0.2
Summary: Yet Another Python Event Emitter
License-Expression: MIT
License-File: LICENSE
Keywords: asyncio,zero-dependency,event-emitter
Author: somespecialone
Author-email: itsme@somespecial.one
Requires-Python: >=3.12
Classifier: Framework :: AsyncIO
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Information Technology
Classifier: Topic :: Software Development
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Project-URL: Bug Tracker, https://github.com/somespecialone/yapee/issues
Project-URL: Homepage, https://github.com/somespecialone/yapee
Project-URL: Repository, https://github.com/somespecialone/yapee
Description-Content-Type: text/markdown

# Yet Another Python Event Emitter

[![Made in Ukraine](https://img.shields.io/badge/made_in-ukraine-ffd700.svg?labelColor=0057b7)](https://stand-with-ukraine.pp.ua)
[![license](https://img.shields.io/github/license/somespecialone/yapee)](https://github.com/somespecialone/yapee/blob/main/LICENSE)
[![Poetry](https://img.shields.io/endpoint?url=https://python-poetry.org/badge/v0.json)](https://python-poetry.org/)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)
[![pypi](https://img.shields.io/pypi/v/yapee)](https://pypi.org/project/yapee)
[![Publish](https://github.com/somespecialone/yapee/actions/workflows/publish.yml/badge.svg)](https://github.com/somespecialone/yapee/actions/workflows/publish.yml)
[![Tests](https://github.com/somespecialone/yapee/actions/workflows/tests.yml/badge.svg)](https://github.com/somespecialone/yapee/actions/workflows/tests.yml)
[![codecov](https://codecov.io/gh/somespecialone/yapee/branch/main/graph/badge.svg?token=B0LARHH9MS)](https://codecov.io/gh/somespecialone/yapee)
[![Ask DeepWiki](https://deepwiki.com/badge.svg)](https://deepwiki.com/somespecialone/yapee)

_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](https://pypi.org) under [yapee](https://pypi.org/project/yapee) name.

```sh
poetry add yapee
```

```sh
pip install yapee
```

```sh
uv add yapee
```

## Usage

### Basic Example

```python
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

```python
@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

```python
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.

```python
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

```python
# 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:

```python
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:

```python
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)
```

