pulq 0.1.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

Quick start

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,
        priority_order=("high", "medium", "low"),
        weights={"high": 3, "medium": 2, "low": 1},
        quantum=1,
    )
    await queue.schedule(Task(priority="high", payload={"job": "a"}))
    await queue.schedule(Task(priority="low", payload={"job": "b"}))

    transport = LocalTransport(queue)
    worker = Worker(transport, "worker-1", handle, no_work_delay_seconds=0.01)

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

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

License

MIT — see LICENSE.