mpyatishev/aioevents

A simple library for managing events through an asynchronous queue

aioevents

A simple library for managing events through an asynchronous queue

Installation

pip install aioevents-ng

Usage example

import asyncio

from dataclasses import dataclass

import aioevents


@dataclass
class MyEvent(aioevents.Event):
   payload: str


@aioevents.manager.register(MyEvent)
async def event_hadler(event: aioevents.Event):
   print(f"recieved: {event}")


async def produce():
   async with aioevents.events as events:
      await events.publish(MyEvent("Hello!"))


async def main():
   aioevents.start(asyncio.get_event_loop())

   await produce()

   print('stopping worker')
   aioevents.stop()

   # wait for all coroutines
   await asyncio.sleep(1)


if __name__ == "__main__":
   asyncio.run(main())

License

aioevents library is offered under Apache 2 license.