pyee is a rough port of node.js’s EventEmitter. It’s missing a number of methods, but implements the basics. Unlike its namesake, it supports wrapping async functions/coroutines as seen in python 3.5.
You can install this project off pypi using
pip install pyee
pyee supplies an
EventEmitter object similar to the
from Node.js. It supports both synchronous callbacks and asyncio coroutines.
In : from pyee import EventEmitter In : ee = EventEmitter() In : @ee.on('event') ...: def event_handler(): ...: print('BANG BANG') ...: In : ee.emit('event') BANG BANG In :
EventEmitter(scheduler=<function ensure_future>, loop=None)¶
The EventEmitter class.
For interoperation with asyncio, one can specify the scheduler and the event loop. The scheduler defaults to
asyncio.ensure_future, and the loop defaults to
None. When used with the default scheduler, this will schedule the coroutine onto asyncio’s default loop.
This should also be compatible with recent versions of twisted by setting
Most events are registered with EventEmitter via the
oncemethods. However, pyee EventEmitters have two special events:
new_listener: Fires whenever a new listener is created. Listeners for this event do not fire upon their own creation.
error: When emitted raises an Exception by default, behavior can be overriden by attaching callback to the event.
@ee.on('error') def onError(message): logging.err(message) ee.emit('error', Exception('something blew up'))
For synchronous callbacks, exceptions are not handled for you— you must catch your own exceptions inside synchronous
onhandlers. However, when wrapping async functions, errors will be intercepted and emitted under the
errorevent. This behavior for async functions is inconsistent with node.js, which unlike this package has no facilities for handling returned Promises from handlers.
emit(event, *args, **kwargs)¶
**kwargsto each attached function. Returns
Trueif any functions are attached to
event; otherwise returns
datais an attached function, this will call
For coroutine event handlers, calling emit is non-blocking. In other words, you do not have to await any results from emit, and the coroutine is scheduled in a fire-and-forget fashion.
Returns the list of all listeners registered to the
Registers the function (or optionally an asyncio coroutine function)
fto the event name
fisn’t provided, this method returns a function that takes
fas a callback; in other words, you can use this method as a decorator, like so:
@ee.on('data') def data_handler(data): print(data)
As mentioned, this method can also take an asyncio coroutine function:
@ee.on('data') async def data_handler(data) await do_async_thing(data)
This will automatically schedule the coroutine using the configured scheduling function (defaults to
asyncio.ensure_future) and the configured event loop (defaults to
The same as
ee.on, except that the listener is automatically removed after being called.
Remove all listeners attached to
None, remove all listeners on all events.
Removes the function