philiprehberger-event-emitter 0.2.3

Type-safe event emitter with sync and async listener support

philiprehberger-event-emitter

Type-safe event emitter with sync and async listener support.

Installation

pip install philiprehberger-event-emitter

Usage

Basic Events

from philiprehberger_event_emitter import EventEmitter

emitter = EventEmitter()

def on_user_created(user):
    print(f"User created: {user['name']}")

emitter.on("user:created", on_user_created)
emitter.emit("user:created", {"name": "Alice"})

Unsubscribe

# Using the returned unsubscribe function
unsubscribe = emitter.on("event", handler)
unsubscribe()

# Or manually
emitter.off("event", handler)

One-Time Listeners

emitter.once("init", lambda: print("Only fires once"))
emitter.emit("init")  # prints
emitter.emit("init")  # nothing

Async Listeners

async def async_handler(data):
    await save_to_db(data)

emitter.on("data:received", async_handler)

# Use async_emit to await async listeners
await emitter.async_emit("data:received", {"key": "value"})

Max Listeners

# Warn when too many listeners are added (helps detect memory leaks)
emitter = EventEmitter(max_listeners=10)

Management

emitter.listener_count("event")      # number of listeners
emitter.event_names()                 # list of events with listeners
emitter.remove_all_listeners("event") # remove all for one event
emitter.remove_all_listeners()        # remove all listeners

API

Method	Description
on(event, listener)	Register listener, returns unsubscribe function
once(event, listener)	Register one-time listener
off(event, listener)	Remove a listener
emit(event, *args, **kwargs)	Emit event synchronously
async_emit(event, *args, **kwargs)	Emit event, awaiting async listeners
listener_count(event)	Count listeners for an event
event_names()	List events with listeners
remove_all_listeners(event?)	Remove all or event-specific listeners

Development

pip install -e .
python -m pytest tests/ -v

License

MIT