pyee

import asyncio from pyee import EventEmitter from pyee.asyncio import AsyncIOEventEmitter # Synchronous EventEmitter sync_ee = EventEmitter() @sync_ee.on('sync_event') def handle_sync_event(data): print(f"Synchronous event received: {data}") sync_ee.emit('sync_event', 'Hello from sync!') # Asynchronous AsyncIOEventEmitter async_ee = AsyncIOEventEmitter() @async_ee.on('async_event') async def handle_async_event(data): print(f"Asynchronous event received: {data}") await asyncio.sleep(0.1) # Simulate async work print("Async handler finished.") @async_ee.on('error') def on_async_error(e): print(f"Async error caught: {e}") async def main(): print("Emitting async event...") async_ee.emit('async_event', 'Hello from async!') # Wait for all handlers to complete (new in v12.1.0) await async_ee.wait_for_complete() print("All async handlers completed.") print("Emitting an error (async)...") async_ee.emit('error', ValueError("Something went wrong in async!")) if __name__ == '__main__': asyncio.run(main())