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. Upgrading? Breaking changes are listed per release in Upgrading.

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