Skip to main content

asynchronous application orchestrator

Project description

It is a small library, which solves single task: orchestration of asynchronous applications.

For example, we have an application, which consists of database, message queue, web API, and background workers. Database and message queue are independent, they can be launched concurrently. Web API and background workers are independent too. But they both depend on database and message queue. So they should be launched after and stopped before database and message queue.

This is how it can be solved using AIOConductor.

from aioconductor import Conductor, Component

class Database(Component):

    async def on_setup(self):
        """ Setup database """

    async def on_shutdown(self):
        """ Shutdown database """

class MessageQueue(Component):

    async def on_setup(self):
        """ Setup message queue """

    async def on_shutdown(self):
        """ Shutdown message queue """

class WebAPI(Component):

    db: Database         # Dependencies are declared by type annotations.
    queue: MessageQueue  # Real instances of the components will be injected
                         # during setup routine.

    async def on_setup(self):
        """ Setup web API """

    async def on_shutdown(self):
        """ Shutdown web API """

class BackgroundWorkers(Component):

    db: Database
    queue: MessageQueue

    async def on_setup(self):
        """ Setup background workers """

    async def on_shutdown(self):
        """ Shutdown background workers """

conductor = Conductor()
conductor.add(WebAPI)
conductor.add(BackgroundWorkers)
conductor.serve()

The code above will run concurrent setup of Database and MessageQueue, and then concurrent setup of WebAPI and BackgroundWorkers. Shutdown will be run in opposite order.

Conductor also provides ability to patch component by alternative implementation. It can be useful for testing.

class MessageQueueMock(Component):
    async def on_setup(self):
        """ Setup message queue mock """

    async def on_shutdown(self):
        """ Shutdown message queue mock """

conductor = Conductor()

conductor.patch(MessageQueue, MessageQueueMock)
# An instance of ``MessageQueueMock`` will be injected into
# ``WebAPI`` and ``BackgroundWorkers``, instead of ``MessageQueue``.

conductor.add(WebAPI)
conductor.add(BackgroundWorkers)
conductor.serve()

Project details


Download files

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

Source Distributions

No source distribution files available for this release.See tutorial on generating distribution archives.

Built Distribution

AIOConductor-0.2-py36-none-any.whl (7.5 kB view details)

Uploaded Python 3.6

File details

Details for the file AIOConductor-0.2-py36-none-any.whl.

File metadata

  • Download URL: AIOConductor-0.2-py36-none-any.whl
  • Upload date:
  • Size: 7.5 kB
  • Tags: Python 3.6
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.4.2 importlib_metadata/4.6.3 pkginfo/1.7.1 requests/2.26.0 requests-toolbelt/0.9.1 tqdm/4.62.0 CPython/3.6.14

File hashes

Hashes for AIOConductor-0.2-py36-none-any.whl
Algorithm Hash digest
SHA256 eff577c600bb3504afc254c685163ecbd5897ca4ae83a734012f9d76299b6b2b
MD5 c580d425241df801b12671fe0320c8b7
BLAKE2b-256 f3564f3297b9f0d3210ab2fd8fdb596c478c5b110aa40ba6df266f4b9fd120ef

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