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.

Source Distribution

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

Uploaded Source

Built Distribution

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

Uploaded Python 3

File details

Details for the file accordian-0.4.0.tar.gz.

File metadata

  • Download URL: accordian-0.4.0.tar.gz
  • Upload date:
  • Size: 3.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.1.1 pkginfo/1.5.0.1 requests/2.23.0 setuptools/41.2.0 requests-toolbelt/0.9.1 tqdm/4.44.1 CPython/3.8.2

File hashes

Hashes for accordian-0.4.0.tar.gz
Algorithm Hash digest
SHA256 2dcf0323d25284171862f750dbbbdef11fdd32b9a8814eaaf5c2f51679b0ed3f
MD5 c0b708384f77b697656b39bd24699362
BLAKE2b-256 dd13f3152ac04e7681cc6131d981565c7d440147a358ed0d1a4c1e7609faf8d3

See more details on using hashes here.

File details

Details for the file accordian-0.4.0-py3-none-any.whl.

File metadata

  • Download URL: accordian-0.4.0-py3-none-any.whl
  • Upload date:
  • Size: 3.8 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.1.1 pkginfo/1.5.0.1 requests/2.23.0 setuptools/41.2.0 requests-toolbelt/0.9.1 tqdm/4.44.1 CPython/3.8.2

File hashes

Hashes for accordian-0.4.0-py3-none-any.whl
Algorithm Hash digest
SHA256 590b40612f59136ecef6968439903f6cee2f34287818473d81b3d524445a86b2
MD5 a26bd099185b40ab8a7fb83138bf39da
BLAKE2b-256 269104c3de96c0ef6d7c3e0088b2eb0ae8357acf483e8a0c6cabfb75bfb8e565

See more details on using hashes here.

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