EventEmitter

The EventEmitter class provides a simple and flexible way to implement an event-driven architecture in Python.

It allows you to register event listeners and emit events to notify those listeners. The emitter supports both synchronous and asynchronous listeners.

Overview

Listeners can be registered using .on(event, listener) and removed using .off(event, listener). To emit events, you can use .emit() (non-blocking) or await .emit_async() (fully awaited).

This pattern is well suited for applications such as:

• Decoupled service layers

• Message/event bus systems

• Plugin architectures

• Real-time user interfaces

Example Usage

from pattern_kit import EventEmitter

def on_data_received(data):
    print(f"[Sync] Got data: {data}")

async def on_data_async(data):
    print(f"[Async] Got data: {data}")

emitter = EventEmitter()
emitter.on("data", on_data_received)
emitter.on("data", on_data_async)

emitter.emit("data", {"value": 42})         # Non-blocking
await emitter.emit_async("data", {"value": 42})  # Awaited

emitter.off("data", on_data_received)

API Reference

class pattern_kit.behavioral.event_emitter.EventEmitter

Bases: object

A lightweight event emitter that supports both synchronous and asynchronous listeners.

emit(event: str, data: Any = None) → None

Emit an event and notify all registered listeners.

Async listeners are scheduled using asyncio.create_task() (non-blocking).

async emit_async(event: str, data: Any = None) → None

Emit an event and await all async listeners (sync ones are called normally).

off(event: str, listener: Callable[[Any], Any]) → None

Unregister a listener from a given event name.

on(event: str, listener: Callable[[Any], Any]) → None

Register a listener for a given event name.