Evently 
A lightweight event bus for python asyncio
About Evently:
- Written in python3.8
- Built around asynio
- Uses type hinting
- Simple API
- Threadsafe
How to install
pip install evently
Dev environment setup
# Clone the repository.
git clone https://github.com/dboslee/evently
cd evently
# Setup virtualenv
pip3 install virtualenv
python3 -m virtualenv env
source env/bin/activate
# Run tests.
python -m unittest discover -p test_*.py
Code Examples
Here is a complete example
import asyncio
from evently import Evently, Event
loop = asyncio.get_event_loop()
evently = Evently(loop)
@evently.subscribe("hello")
async def say_world(event: Event):
print("world")
async def main():
evently.publish("hello")
await asyncio.sleep(0)
loop.run_until_complete(main())
Pass data along with an event and access it with event.event_data
@evently.subscribe("hello")
def say_world(event):
print(event.event_type)
print(event.event_data)
print("world")
evently.publish("hello", {"example": "data"})
Subscribe to the same event many times
@evently.subscribe("hello")
def say_world(event):
print("world")
@evently.subscribe("hello")
def say_goodbye(event):
print("goodbye")
Subscribe to multiple events
@evently.subscribe(["hello", "hola"])
def say_world(event):
if event.event_type == "hello":
print("world")
elif event.event_type == "hola":
print("mundo")
Unsubscribe when you no longer want a handler to be called
@evently.subscribe("hello")
def say_world(event):
print("world")
say_world.unsubscribe()
Run blocking code in an executor with the kwarg blocking=True
from time import sleep
@evently.subscribe("hello", blocking=True)
def say_world_blocking(event):
sleep(10)
print("world")
Receive the EventHandler instance as a kwarg by using context=True
from time import sleep
@evently.subscribe("hello", context=True)
def say_world_blocking(event, context):
sleep(10)
print("world")
context.unsubscribe()
