toro 🐂

An async-first, Redis-backed job queue for Python. Every state transition is an atomic Lua script; producing and processing are asyncio end to end.

pip install toro-queue      # the import name is `toro`

Installed as toro-queue on PyPI (the name toro was taken), but you import toro. See the docs for the architecture, the reliability model, and the detailed guides.

Pairs with matador, a live web dashboard for your queues.

Why toro

• Async-native. Enqueue and process with async/await - no thread pools, no sync bridge. A natural fit for FastAPI, aiohttp, or any asyncio app.
• Atomic by construction. Claims, retries, promotions and finishes are Lua scripts, so a job can't be lost or double-committed between two round trips.
• At-least-once delivery. Per-job locks + a background mark-and-sweep recover jobs from workers that crashed - without the visibility-timeout double-delivery trap of some other queues.
• Typed. Ships py.typed; the public API is fully annotated.

Features

Quick start

import asyncio
from toro import Queue, Worker

async def main():
    queue = Queue("emails")
    await queue.add("welcome", {"to": "ada@example.com"})

    async def process(job):
        print("sending", job.data)
        return {"ok": True}

    worker = Worker("emails", process, concurrency=8)
    worker.on("completed", lambda job, result: print("done", job.id))
    await worker.run()

asyncio.run(main())

A taste of the options

# Priorities, delay, and retry-with-backoff
await queue.add("report", data, priority=10, delay=5000,
                attempts=5, backoff={"type": "exponential", "delay": 1000})

# Idempotent custom id (a second add with the same id is ignored)
await queue.add("charge", data, job_id="order-1234")

# A repeatable schedule (cron or every-N-ms); "run now" with trigger_scheduler
await queue.add_scheduler("nightly-rollup", cron="0 0 * * *")

# A flow: children run first (fan-out), the parent runs on their results (fan-in)
from toro import FlowChild as c
report = await queue.add_flow("report", {"q": 3},
                              children=[c("fetch", {"shard": i}) for i in range(3)])

# Queue-wide rate limit: at most 100 jobs / second across every worker
worker = Worker("emails", process, rate_limit={"max": 100, "duration": 1000})

# Wait for a result from the producer side
job = await queue.add("resize", {"src": "a.png"})
print(await job.result(timeout=30))

Flows

A flow enqueues a parent and its children as one atomic tree. The children run first (fan-out, nested arbitrarily); the parent parks until every child has settled, then runs and reads their results (fan-in). One primitive covers fan-out/fan-in and chained steps, with per-child failure policies and flow-aware retry that recovers a whole failed flow in one shot. Full guide: docs/flows.md.

Develop

Managed with uv; the Astral toolchain throughout.

uv sync                          # venv + deps + dev group
uv run ruff check .              # lint  (strict: select = ALL)
uv run ruff format .             # format
uv run ty check                  # type check
uv run pytest -m "unit or integration"   # tests (integration needs Redis on :6379)
uv run python examples/basic.py

The suite is a pyramid - -m unit (fast, no Redis), -m integration (Redis), and -m load (the open-loop benchmark harness in tests/load/).

License

MIT