pulq 0.2.0

Fair task scheduling with async pull: WDRR-based priority queue for Python

PULQ

Fair task scheduling with async pull — WDRR-based priority queue for Python

PULQ schedules pull-based work across named priority buckets using Weighted Deficit Round Robin (WDRR) so higher-weight classes get proportionally more CPU time without starving lower classes. Management commands (e.g. STOP) are delivered per worker ahead of normal tasks.

Install

pip install pulq

For HTTP client transport: pip install pulq[http] — see docs/transports.md.

Quick start

PullQueue and Worker use sensible defaults so you can schedule tasks and run a worker with almost no configuration:

import asyncio

from pulq import (
    CommandType,
    InMemoryTaskRepository,
    LocalTransport,
    PullQueue,
    Task,
    Worker,
)

async def handle(task: Task) -> dict:
    return {"ok": True, "echo": task.payload}

async def main() -> None:
    repo = InMemoryTaskRepository()
    queue = PullQueue(repo)  # default: high / medium / low with 3:2:1 weights, quantum 1
    await queue.schedule(Task(priority="high", handler_name="default", payload={"job": "a"}))
    await queue.schedule(Task(priority="low", handler_name="default", payload={"job": "b"}))

    transport = LocalTransport(queue)
    worker = Worker(transport, "worker-1", handle)  # default short backoff when idle

    async def stop_later() -> None:
        await asyncio.sleep(0.05)
        queue.send_command("worker-1", CommandType.STOP)

    await asyncio.gather(worker.run(), stop_later())

asyncio.run(main())

Customizing with Pydantic config

For different priorities, weights, quantum, or worker lifecycle hooks on the handler registry, pass validated config objects:

from pulq import DeficitSchedulerConfig, HandlerRegistry, PullQueueConfig, WorkerConfig

queue = PullQueue(
    repo,
    config=PullQueueConfig(
        scheduler=DeficitSchedulerConfig(
            priority_order=("critical", "high", "low"),
            weights={"critical": 5, "high": 3, "low": 1},
            quantum=2,
        ),
    ),
)

registry = HandlerRegistry(default=handle, startup=my_startup, shutdown=my_shutdown)
worker = Worker(
    transport,
    "worker-1",
    registry,
    config=WorkerConfig(no_work_delay_seconds=0.05),
)

More on HTTP vs local and repositories: docs/transports.md.

Documentation

Full docs: pulq.readthedocs.io

Development

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
ruff check src tests
mypy src
pytest

Releases (PyPI + Read the Docs): after bumping the version in pyproject.toml and updating CHANGELOG.md, merge to main — see docs/releasing.md for trusted publishing setup on PyPI and manual options.

License

MIT — see LICENSE.