taskq

A small job queue you can read in an afternoon. Submit work over HTTP, run it on worker processes, and watch it move on a live dashboard.

Built with FastAPI, Redis, PostgreSQL and asyncio. It covers the parts of Celery and RQ most projects actually use — retries with backoff, scheduled jobs, a dead-letter queue, graceful shutdown — in about 1,500 lines of Python that fit in your head.

┌─ POST /api/jobs ──► PostgreSQL (the record) ──► Redis (the queue) ──► worker
│                                                                         │
└──────────── dashboard ◄── GET /api/stats ◄── job row updated ◄──────────┘

Why it exists

Celery is powerful and enormous. RQ is simpler but synchronous and Redis-only. taskq sits between them: async from top to bottom, with job history in a real database so you can query it, and a dashboard that shows what is happening right now rather than what happened when you last refreshed.

It is a learning project with production manners: every failure mode below is covered by a test.

Quick start

git clone https://github.com/yourname/taskq && cd taskq
make up          # Postgres, Redis, the API, two workers and the scheduler
make demo        # submit a burst of jobs to watch
open http://localhost:8000

Or without Docker:

pip install -e ".[dev]"
cp .env.example .env         # point it at your own Redis and Postgres

taskq init-db
taskq api                    # http://localhost:8000
taskq worker -c 4            # in a second terminal
taskq scheduler              # in a third

Submit something:

curl -X POST localhost:8000/api/jobs \
  -H 'Content-Type: application/json' \
  -d '{"task": "taskq.tasks.add", "args": [2, 3]}'

{ "id": "3f0c…", "task": "taskq.tasks.add", "state": "pending", "attempts": 0, … }

Writing tasks

from taskq import task


@task(max_retries=5, timeout=30)
async def send_welcome_email(user_id: int) -> dict:
    """Anything awaitable works, and plain functions run in a thread."""
    user = await db.fetch_user(user_id)
    await mailer.send(user.email, template="welcome")
    return {"sent_to": user.email}

Point TASKQ_TASK_MODULES at the module holding them so the API and workers both import it. Enqueue from your own code without going through HTTP:

from taskq.client import TaskQueue

queue = TaskQueue()
await queue.enqueue("myapp.tasks.send_welcome_email", kwargs={"user_id": 7})
await queue.enqueue("myapp.tasks.nightly_report", delay=3600)  # in an hour
await queue.enqueue("myapp.tasks.charge", run_at=renewal_date)  # at a time

How a job travels

stateDiagram-v2
    [*] --> pending: submitted
    pending --> running: claimed by a worker
    running --> completed: returned
    running --> failed: raised, retries left
    failed --> running: backoff elapsed
    running --> dead: retries spent
    dead --> pending: requeued from the dashboard
    pending --> cancelled: withdrawn

Loading

Four Redis structures carry it:

Key	Type	Holds
taskq:ready:<queue>	list	ids waiting for a worker, FIFO
taskq:delayed	zset	ids scored by the second they come due
taskq:inflight:<worker>	list	ids one worker has claimed
taskq:dead	list	ids that exhausted their retries

PostgreSQL holds the payload, the result, the traceback and the timings. Redis only ever holds ids, so a job's history survives a FLUSHALL.

The guarantees, and the reasoning behind them

A job is never lost between the queue and a worker. Claiming is a single LMOVE from ready into that worker's inflight list, so the id is in exactly one place at every instant.

A crashed worker does not take its jobs with it. Each worker refreshes a heartbeat key with a TTL. When the key expires, the scheduler's reaper drains that worker's inflight list back onto the ready queue and resets the rows to pending.

A job is never queued before it is recorded. submit() commits the database row first and publishes to Redis second. The reverse order would let a worker claim an id the database has not heard of. The remaining gap — a crash between the two — is closed by recover_orphans(), which runs at scheduler startup and re-publishes any waiting row that Redis does not know about.

Delivery is at-least-once. A worker that dies mid-job will have that job run again, so tasks should be idempotent. Exactly-once would need a two-phase commit across Redis and Postgres, which is not a trade this project makes.

Retries

A failed job goes back into the delayed set rather than onto the ready queue, so it waits its turn:

delay = backoff_base × backoff_factor ^ (attempts - 1)   capped at backoff_max

With the defaults that is 2s, 4s, 8s, 16s… up to ten minutes, plus ±25% jitter so a thousand jobs failing on the same outage don't return in lockstep. Set any of it per task or per job:

curl -X POST localhost:8000/api/jobs -H 'Content-Type: application/json' -d '{
  "task": "myapp.tasks.call_flaky_api",
  "max_retries": 8,
  "backoff_base": 5,
  "backoff_max": 900
}'

When the retries run out the job is marked dead and parked in the dead-letter queue, where it waits for a human. Requeue one from the drawer in the dashboard, or the lot with POST /api/dead-letter/requeue.

Graceful shutdown

On SIGTERM or SIGINT a worker stops claiming, lets its running jobs finish within TASKQ_SHUTDOWN_GRACE_PERIOD, then hands back anything still unfinished so another worker picks it up. An interrupted attempt is not counted against the job's retry budget — it never got a fair run. Give containers a stop_grace_period longer than the setting, or the platform will kill what the worker is trying to drain.

Dashboard

http://localhost:8000 — throughput for the last hour with completed above the line and failed below, live worker cards, queue depths, and a job table that filters by state. Click any row for arguments, result, traceback, and buttons to retry, cancel or delete. It refreshes every two seconds and pauses when the tab is hidden.

API

Method	Path	Does
POST	/api/jobs	Submit a job, now or scheduled
GET	/api/jobs	List jobs; filter by state, queue, task
GET	/api/jobs/{id}	One job in full
POST	/api/jobs/{id}/cancel	Withdraw a job that has not started
POST	/api/jobs/{id}/retry	Requeue a dead, failed or cancelled job
DELETE	/api/jobs/{id}	Delete the record and any queue entry
GET	/api/dead-letter	What is parked, and why
POST	/api/dead-letter/requeue	Send it all back
DELETE	/api/dead-letter	Clear it
GET	/api/stats	Everything the dashboard draws
GET	/api/queues · /api/workers · /api/tasks	Depths, workers, registered tasks
GET	/health	Liveness, including Redis

Interactive docs at /docs.

Priority queues

Workers drain queues in the order they are given, so the first name wins while it has work:

taskq worker -q urgent -q default

Configuration

Every setting is an environment variable prefixed TASKQ_, or a line in .env. See .env.example for the full list with defaults. The ones worth knowing:

Variable	Default	Meaning
TASKQ_DATABASE_URL	local Postgres	SQLAlchemy async URL
TASKQ_REDIS_URL	local Redis	Where the queues live
TASKQ_QUEUES	["default"]	Queues a worker drains, in priority order
TASKQ_WORKER_CONCURRENCY	4	Jobs one worker runs at once
TASKQ_JOB_TIMEOUT	300	Seconds before a job is cut short
TASKQ_SHUTDOWN_GRACE_PERIOD	30	Seconds to drain on SIGTERM
TASKQ_MAX_RETRIES	3	Attempts after the first
TASKQ_HEARTBEAT_TTL	30	Silence before a worker is presumed dead

Command line

taskq api          Serve the REST API and dashboard
taskq worker       Run a worker   (-q queue, -c concurrency)
taskq scheduler    Promote due jobs and reap lost workers
taskq submit       Enqueue a job from the shell
taskq stats        Print a snapshot of the queue
taskq init-db      Create the schema

Tests

make test     # 59 tests, about four seconds
make cov

The suite runs against SQLite and fakeredis, so it needs no services and no network. It covers the interesting failures rather than only the happy path: a worker killed mid-job, a job that outlives its timeout, retries landing in the right order, the reaper rescuing work from a silent worker, and Redis losing a queue that Postgres still remembers.

CI also runs the whole stack against real Redis and Postgres, submits a job, and checks that a SIGTERM drains rather than kills.

What it does not do

Worth knowing before you reach for it:

• No task chaining or workflows. No chains, groups or chords. A task that needs to trigger another enqueues it itself.
• No result backend beyond Postgres. Results are stored as JSON on the job row; anything not JSON-serialisable is stored as its repr.
• No rate limiting or worker autoscaling.
• No schema migrations. create_schema() runs at startup. Add Alembic before you change a column in anger.
• At-least-once, not exactly-once. Keep tasks idempotent.

Layout

taskq/
├── broker.py      Redis: ready, delayed, in-flight, dead-letter, heartbeats
├── worker.py      Claim, run, retry, drain on shutdown
├── scheduler.py   Promote due jobs, reap lost workers, recover orphans
├── models.py      The job row, its states, the backoff maths
├── client.py      submit() and TaskQueue
├── registry.py    The @task decorator
├── cli.py         taskq …
├── api/           FastAPI routes
└── static/        The dashboard

Licence

MIT. See LICENSE.