Skip to main content

Event dispatch in Python 3.8 using asyncio

Project description

Event dispatch in Python 3.8 using asyncio


pip install accordian

Getting Started

import asyncio
from accordian import signal
my_event = signal("my_event")

async def pow(x, y):
    await asyncio.sleep(0.1)
    print(f"pow({x}, {y})")
    return x ** y

async def diff(x, y):
    await asyncio.sleep(0.2)
    print(f"diff({x}, {y})")
    return max(x, y) - min(x, y)

# create events without blocking
async def create():
    my_event.send(4, 6)
    my_event.send(0, 1)

# or block and collect results from all receivers
async def collect():
    results = await my_event.join(10, 3)
    assert set(results) == {1000, 7}

# signals without receivers return no results
async def empty():
    sig = signal("unknown")
    results = await sig.join(1, "foo", None)
    assert results == []


By default, accordian.signal creates signals in a global namespace. You can create your own namespaces to group signals together. Here, a processor is passed the region and stage to create deployment tasks:

from accordian import Namespace
regions = {"east": Namespace(), "west": Namespace()}

async def deploy_east_dev(s3_url, creds):

async def deploy_east_prod(s3_url, creds):
    # remove pre-prod feature flags
    await sanitize_prod(s3_url, "east")

async def deploy_west_prod(s3_url, creds):
    # legacy region shims
    await patch_west_bundle(s3_url)
    await sanitize_prod(s3_url, "west")

async def deploy(region, stage):
    s3_url = await bundle_for_region(region, stage)
    creds = await creds_for_region(region, stage)
    signal = regions[region].signal(stage)

    # create the deployment task without waiting
    signal.send(s3_url, creds)

# create deployment tasks"east", "dev"))"west", "prod"))

# wait for deployments to complete
async def wait_for_tasks():
    running = asyncio.all_tasks()
    await asyncio.wait(running)


Contributions welcome! Please make sure tox passes before submitting a PR.


To set up a virtualenv and run the test suite:

git clone
make venv

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

accordian-0.4.0.tar.gz (3.2 kB view hashes)

Uploaded Source

Built Distribution

accordian-0.4.0-py3-none-any.whl (3.8 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page