Python background queues with an async twist

Use async functions to manage complex background task workflows, and keep using synchronous functions for everything else.

Getting Started

Install queueio:

pip install queueio

Create your routines:

# basic.py
from time import sleep

from queueio import gather
from queueio import pause
from queueio import routine


@routine(name="blocking", queue="basic")
def blocking():
    sleep(0.1)  # Regular blocking call


@routine(name="yielding", queue="basic")
async def yielding(iterations: int):
    # Do them two at a time
    for _ in range(iterations // 2):
        await gather(blocking(), blocking())
        await pause(0.2)  # Release processing capacity
    if iterations % 2 == 1:
        await blocking()

Add the configuration to your pyproject.toml:

[tool.queueio]
# Configure the broker (currently supports amqp:// URIs)
broker = "amqp://guest:guest@localhost:5672/"
# Register the modules that the worker should load to find your routines
register = ["basic"]

The broker configuration can be overridden with an environment variable to allow a project to be deployed in multiple environments.

QUEUEIO_BROKER='amqp://guest:guest@localhost:5672/'

QueueIO requires RabbitMQ's global QoS for shared concurrency limits and suspending tasks. On RabbitMQ 4.3, enable it in rabbitmq.conf and restart the broker:

deprecated_features.permit.global_qos = true

Sync the queues to the broker:

queueio sync

Submit the routine to run on a worker:

from queueio import activate
from basic import yielding

with activate():
    yielding(7).submit()

Then run the worker to process submitted routines:

queueio run basic=4

Monitor the status of active routine invocations:

queueio monitor

Stability

The design of the public API is under active development and is likely to change with any release. Release notes will provide upgrade instructions, but backward compatibility and deprecation warnings will not generally be implemented.