Skip to main content

Event dispatch in Python 3.8 using asyncio

Project description

https://img.shields.io/travis/numberoverzero/accordian/master.svg?style=flat-square https://img.shields.io/pypi/v/accordian.svg?style=flat-square https://img.shields.io/github/issues-raw/numberoverzero/accordian.svg?style=flat-square

Event dispatch in Python 3.8 using asyncio

Installation

pip install accordian

Getting Started

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

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

@my_event.connect
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)
asyncio.run(create())


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


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

Namespaces

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()}


@regions["east"].signal("dev").connect
async def deploy_east_dev(s3_url, creds):
    ...

@regions["east"].signal("prod").connect
async def deploy_east_prod(s3_url, creds):
    # remove pre-prod feature flags
    await sanitize_prod(s3_url, "east")
    ...

@regions["west"].signal("prod").connect
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
asyncio.run(deploy("east", "dev"))
asyncio.run(deploy("west", "prod"))


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

Contributing

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

Development

To set up a virtualenv and run the test suite:

git clone https://github.com/numberoverzero/accordian.git
make venv
make

Project details


Download files

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

Files for accordian, version 0.4.0
Filename, size File type Python version Upload date Hashes
Filename, size accordian-0.4.0-py3-none-any.whl (3.8 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size accordian-0.4.0.tar.gz (3.2 kB) File type Source Python version None Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page