Distributed Job Queue with FastAPI

Build a Redis-backed job queue with retries, dead letters, and auto-scaling.

Introduction

30 Second Summary

Every time you upload a photo and it gets resized in the background, or place an order and receive a confirmation email seconds later, a job queue is handling the work behind the scenes. These invisible pipelines keep apps responsive by moving slow tasks off the main request.

In this project, you will build a distributed job queue system from scratch using FastAPI and Redis. You will submit jobs through a REST API, process them with async workers, and handle failures with a Dead Letter Queue.

What You'll Build

You will fire off 50 jobs through a single API call and watch your terminal light up as workers multiply from 1 to 10, processing the backlog in parallel before quietly scaling back down.

By the end of this project, you'll have:

  • A REST API that submits jobs to a Redis-backed queue and tracks each job's status in real time.
  • Retry logic with exponential backoff that re-attempts failed jobs automatically, plus a Dead Letter Queue where permanently failed jobs land for inspection and replay.
  • An auto-scaling worker pool that grows from 1 to 10 concurrent tasks when queue depth spikes, then shrinks back down as the backlog clears.
  • Secret Mission: Add priority-based routing with high, medium, and low queues so urgent jobs always process first.

Are there any prerequisites?

No prerequisites are needed. This project walks you through setting up Python, Docker, and Redis from scratch in Step 1.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Your Environment

Every distributed system depends on infrastructure before you can write any business logic. Your job queue needs a Redis server to store and deliver jobs, and Python packages to build the API that talks to it.

In this step, you'll verify your local tools, spin up Redis inside a Docker container, and create a Python project with the right dependencies installed.

In this step, get ready to:
  • Verify Python, VS Code, and Docker Desktop are ready.
  • Start a Redis server inside a Docker container.
  • Create a project folder with a virtual environment and install dependencies.
Verify your tools are ready

You already have Python 3.14, VS Code, and Docker Desktop installed. Let's confirm each one is ready before moving on.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • Open a terminal inside VS Code by pressing Ctrl+` (backtick key, top-left of keyboard).
  • Verify your Python version by running this command:
python --version

You should see Python 3.14 followed by a patch number (for example, Python 3.14.7).

Not seeing Python 3.14?

If you see an older version, try running python3 --version instead. On some systems, python points to an older installation.

If neither command shows 3.14, download and install Python 3.14 from the official Python downloads page.

Help me troubleshoot my Python installation.

Now make sure Docker Desktop is running so you can start containers.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Docker Desktop and press Enter to open it.
  • Wait until Docker Desktop shows a green status indicator in the bottom-left corner confirming the engine is running.

Docker Desktop not starting?

Docker Desktop can take 30-60 seconds to start up the first time. If it stays stuck, try restarting Docker Desktop from the system tray icon.

If Docker Desktop is not installed, download it from the official Docker Desktop page.

Help me get Docker Desktop running on my machine.

Start Redis in a Docker container

Redis is an in-memory data store that your job queue will use as its backbone. Instead of installing Redis directly on your machine, you'll run it inside a Docker container for clean isolation.

  • Start a Redis container by running this command in your VS Code terminal:
docker run -d --name redis-dev -p 6379:6379 redis:8

What does this command do?

  • -d runs the container in the background so your terminal stays free.
  • --name redis-dev gives the container a name you can reference later.
  • -p 6379:6379 maps Redis's default port 6379 to the same port on your machine.
  • redis:8 pulls the official Redis 8 image from Docker Hub.

Docker will download the Redis image the first time you run this. This may take a minute depending on your internet speed.

  • Verify Redis is running by running this command:
docker exec redis-dev redis-cli ping

You should see PONG in your terminal. This confirms Redis is running and accepting connections.

Not seeing PONG?

If you see an error about the container name already being in use, a previous redis-dev container may exist from an earlier attempt. Remove it first, then try again:

docker rm -f redis-dev
docker run -d --name redis-dev -p 6379:6379 redis:8

If you see an error about the Docker daemon not running, make sure Docker Desktop is fully started with a green status icon.

Help me troubleshoot my Redis Docker container.

Create your project folder and install dependencies

With Redis running, you need a Python project that can talk to it. You'll create a dedicated folder with a virtual environment to keep your dependencies isolated.

  • Move to your Desktop and create your project folder by running these commands:
cd ~/Desktop
mkdir job-queue
cd job-queue
  • Create a Python virtual environment inside this folder by running:
python -m venv venv
  • Confirm the virtual environment was created by running:
ls

You should see a venv folder listed in the output.

  • Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

Seeing a script execution error?

If you see an error saying that running scripts is disabled on this system, run the following command first, then try activating again:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

Linux

source venv/bin/activate

Your terminal prompt now shows (venv) at the beginning. This confirms the virtual environment is active.

What is a virtual environment?

A virtual environment is an isolated Python setup inside your project folder. Any packages you install go into this environment, not your system-wide Python. This prevents version conflicts between different projects on your machine.

  • Install the project dependencies by running:
pip install fastapi==0.141.1 uvicorn==0.52.3 redis==8.1.0

What are these packages?

  • fastapi is the web framework you'll use to build your job queue's REST API endpoints.
  • uvicorn is the ASGI server that runs FastAPI and handles incoming HTTP requests.
  • redis is the Python client for talking to Redis. It includes async support via redis.asyncio, which lets your app interact with Redis without blocking.

You should see output confirming all three packages were installed along with their dependencies.

Before you run the next command, what do you think it will print if all three packages installed correctly?

  • Test that everything is wired up by running:
python -c "import fastapi; import redis.asyncio; print('ready')"

You should see ready printed in your terminal. This confirms FastAPI and redis-py (including its async module) are all importable.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is still active. You should see (venv) at the start of your terminal prompt. If it disappeared, re-activate it and run the pip install command again.

If you used python3 earlier to check your version, use it here too:

python3 -c "import fastapi; import redis.asyncio; print('ready')"

Help me fix my Python imports for FastAPI and redis.

Your environment is locked and loaded. Next up, you'll build the core job queue and submit your first job through a FastAPI endpoint.

Build the Job Queue and Submit Your First Job

Your environment is set up and Redis is running in Docker. Now it is time to build the core of your job queue system.

A job queue is only useful if you can put something in and take something out. This step builds that minimum viable loop: an API endpoint that pushes a job into Redis, and a worker that pops it off and processes it.

In this step, get ready to:
  • Create a FastAPI app with a Redis connection and job submission endpoints.
  • Add a background worker that pops jobs from the queue and processes them.
  • Add a stats endpoint to inspect queue depth.
Create your FastAPI app and job endpoints

Every job queue needs two sides: a producer that pushes jobs in, and a consumer that pulls them out. You will start with the producer side by creating a FastAPI app that connects to Redis on startup and exposes an endpoint to submit jobs.

  • In VS Code, create a new file called main.py inside your project folder.
  • Add the app skeleton with imports, Redis connection config, and a lifespan that connects to Redis on startup:
import asyncio
import json
import time
import uuid

import redis.asyncio as aioredis
from fastapi import FastAPI
from contextlib import asynccontextmanager

REDIS_URL = "redis://localhost:6379"

# Global Redis client, set during startup
redis_client: aioredis.Redis = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Connect to Redis and verify the connection
    global redis_client
    redis_client = aioredis.from_url(REDIS_URL, decode_responses=True)
    await redis_client.ping()
    print("[Startup] Connected to Redis")

    yield

    await redis_client.aclose()
    print("[Shutdown] Cleaned up")

app = FastAPI(title="Distributed Job Queue", lifespan=lifespan)

What does this code do?

The imports bring in everything you need: asyncio for async workers later, json to serialize jobs, uuid for unique job IDs, and redis.asyncio for async Redis operations.

The lifespan function is FastAPI's recommended pattern for startup and shutdown logic. Before yield, it connects to Redis and verifies the connection with ping(). After yield (when the app shuts down), it closes the Redis connection cleanly.

  • Save main.py.
  • Start the app by running this command in your terminal (make sure your virtual environment is active):
uvicorn main:app --reload

You should see output including [Startup] Connected to Redis and the server running at http://127.0.0.1:8000.

What does this command do?

This tells uvicorn to run the app object from main.py. The --reload flag watches your code for changes and automatically restarts the server when you save.

Seeing a connection error on startup?

If the app fails to start with a connection error, your Redis container might not be running. Open a new terminal and run docker ps. You should see redis-dev listed. If not, restart it by running docker start redis-dev.

Help me debug my Redis connection

  • Now add the POST /jobs endpoint below the app = FastAPI(...) line in main.py:
@app.post("/jobs")
async def create_job(payload: dict | None = None):
    # Generate a unique ID for this job
    job_id = str(uuid.uuid4())
    job = {
        "id": job_id,
        "payload": payload or {},
        "created_at": str(time.time()),
    }
    # Store initial status in a Redis hash
    await redis_client.hset(
        f"job:{job_id}", mapping={"status": "queued"}
    )
    # Push the job onto the queue list
    await redis_client.lpush("queue:jobs", json.dumps(job))
    return {"job_id": job_id, "status": "queued"}

What does this code do?

When someone POSTs to /jobs, this endpoint generates a unique ID with uuid4() and builds a job dictionary.

It then does two things in Redis. First, hset creates a hash at the key job:{job_id} to track the job's status. Second, LPUSH pushes the full job payload onto a Redis list called queue:jobs. That list is your queue.

  • Save main.py. The server auto-reloads.
  • Open your browser and go to http://127.0.0.1:8000/docs.
  • Click on the POST /jobs row to expand it.
  • Click Try it out.
  • Leave the request body as-is (or type {}) and click Execute.

You should see a JSON response with a job_id and "status": "queued". Your job is now sitting in the Redis queue waiting to be processed.

What is this page?

FastAPI automatically generates interactive API docs at /docs using Swagger UI. You can test every endpoint directly from this page without needing a separate tool.

Not seeing the Swagger UI?

Make sure uvicorn is still running in your terminal. Check that you saved main.py before starting the server.

If you see a connection error, check that the Redis container is running by running docker ps in a new terminal. You should see redis-dev listed.

Help me debug the Swagger UI issue

  • Now add the GET /jobs/{job_id} endpoint below your create_job function in main.py:
@app.get("/jobs/{job_id}")
async def get_job_status(job_id: str):
    # Look up the job's status hash in Redis
    data = await redis_client.hgetall(f"job:{job_id}")
    if not data:
        return {"error": "Job not found"}
    return {"job_id": job_id, **data}

What does this code do?

This endpoint reads the status hash you stored earlier with hgetall. You pass in a job ID, and Redis returns all the fields stored in that hash. Right now, the only field is status: queued. The worker you build next will update this hash as it processes jobs.

  • Save main.py. The server auto-reloads.
  • Back in the Swagger UI at http://127.0.0.1:8000/docs, refresh the page.
  • Expand GET /jobs/{job_id}.
  • Click Try it out.
  • Paste the job_id from your earlier POST response into the job_id field.
  • Click Execute.

You should see "status": "queued". The job is still sitting in the queue because nothing is consuming it yet. Time to fix that.

Add a background worker to process jobs

Your queue has a producer (the /jobs endpoint) but no consumer. Jobs go in and never come out. The worker is an async loop that waits for jobs to appear in the Redis list, pops them off, and marks them as completed.

  • In main.py, add the worker_loop function above the lifespan function (it must be defined before lifespan references it):
async def worker_loop(name: str):
    global redis_client
    try:
        while True:
            # Block-wait up to 1s for a job
            result = await redis_client.brpop("queue:jobs", timeout=1)
            if result is None:
                continue
            _, raw = result
            job = json.loads(raw)
            job_id = job["id"]

            # Mark job as processing
            await redis_client.hset(
                f"job:{job_id}",
                mapping={"status": "processing", "worker": name},
            )

            # Simulate work
            await asyncio.sleep(0.5)

            # Mark job as completed
            await redis_client.hset(
                f"job:{job_id}",
                mapping={
                    "status": "completed",
                    "worker": name,
                    "completed_at": str(time.time()),
                },
            )
            print(f"[{name}] Processed job {job_id}: completed")

    except asyncio.CancelledError:
        print(f"[{name}] Shutting down")

What does this code do?

The worker sits in an infinite loop calling BRPOP on the queue:jobs list. BRPOP is a blocking pop that waits up to 1 second for a new item to appear. This is efficient because the worker uses no CPU while the queue is empty.

When a job arrives, the worker updates its status to processing, simulates work with a half-second sleep, then marks it completed. The CancelledError handler ensures the worker shuts down cleanly when the app stops.

Why BRPOP instead of polling?

You could check the queue with LLEN and then pop with RPOP in a loop. That approach burns CPU cycles checking an empty queue over and over. BRPOP hands that waiting off to Redis itself, which is designed for exactly this kind of blocking wait.

Now you need to wire the worker into the app's lifespan so it starts automatically.

  • In main.py, find this section inside the lifespan function:
    print("[Startup] Connected to Redis")

    yield

    await redis_client.aclose()
  • Replace it with this, which creates a worker task before yield and cancels it on shutdown:
    print("[Startup] Connected to Redis")

    worker_task = asyncio.create_task(worker_loop("Worker-1"))

    yield

    worker_task.cancel()
    await redis_client.aclose()

What changed?

You added two lines. asyncio.create_task(worker_loop("Worker-1")) launches the worker as a background coroutine that runs alongside your API. worker_task.cancel() stops it cleanly when the server shuts down.

  • Save main.py. The server auto-reloads.

Before you submit a job, think about what will happen differently this time. Earlier, your job sat in the queue with a status of queued forever. What do you expect to see now that a worker is running?

  • In the Swagger UI, expand POST /jobs.
  • Click Try it out.
  • Click Execute.
  • Switch back to the terminal where uvicorn is running.

You should see a log line like [Worker-1] Processed job <your-job-id>: completed. The job traveled from your HTTP request, through the Redis queue, to the worker.

  • Back in the Swagger UI, expand GET /jobs/{job_id}.
  • Click Try it out.
  • Paste the job_id from the POST response into the job_id field.
  • Click Execute.

The status should now read "status": "completed" with a worker field showing Worker-1.

Job status still showing queued?

Check the terminal for error messages. If you see an import error, make sure worker_loop is defined above the lifespan function in your file. Python needs to see the function definition before it is referenced.

If the terminal shows [Startup] Connected to Redis but no worker log, check that you added the asyncio.create_task line before yield in the lifespan function, not after it.

Help me debug my worker

Add a queue stats endpoint

You can submit and process jobs, but you have no way to see how many jobs are waiting. A stats endpoint gives you a window into the queue's state.

  • Add the GET /queue/stats endpoint below your get_job_status function in main.py:
@app.get("/queue/stats")
async def queue_stats():
    # LLEN returns the number of items in the Redis list
    queue_depth = await redis_client.llen("queue:jobs")
    return {
        "queue_depth": queue_depth,
    }

What does this code do?

Redis LLEN returns the length of a list. By checking queue:jobs, you can see how many jobs are waiting to be processed. Since your worker processes jobs almost instantly, you will usually see 0 here. That changes in later steps when you add failure simulation and batch submissions.

  • Save main.py. The server auto-reloads.
  • Refresh the Swagger UI page.
  • Expand GET /queue/stats.
  • Click Try it out.
  • Click Execute.

You should see "queue_depth": 0. The worker already consumed your job, so nothing is waiting.

Seeing queue_depth greater than 0?

If you submitted jobs before the worker was wired up, those old jobs may still be sitting in the queue. Stop the server with Ctrl+C, then restart it with the same uvicorn command. The worker will drain the leftover jobs on startup.

Help me debug unexpected queue stats

✔️ Awesome, I've got everything!

Great. Make sure you have saved main.py and your server is running.

ⓧ I'd like to double check the full code

Scroll up and confirm your main.py contains each of these sections in order:

  • Imports (asyncio, json, time, uuid, redis.asyncio, FastAPI, asynccontextmanager).
  • REDIS_URL constant and redis_client global.
  • worker_loop function using BRPOP with a 1-second timeout.
  • lifespan function that connects to Redis, creates a Worker-1 task, and cancels it on shutdown.
  • app = FastAPI(...) with the lifespan parameter.
  • POST /jobs endpoint with LPUSH.
  • GET /jobs/{job_id} endpoint with hgetall.
  • GET /queue/stats endpoint with LLEN.

Your job queue loop is complete. You can submit jobs, they flow through Redis, and a worker picks them up and marks them done. Next up, you will add failure simulation and retry logic so your queue can handle jobs that do not succeed on the first try.

Add Retry Logic with Exponential Backoff

Your job queue processes every job successfully right now. In production, that never happens. Network calls time out, APIs return errors, and services crash mid-request.

In this step, you will simulate random failures and watch jobs vanish without a trace. Then you will fix the problem with retry logic that automatically re-enqueues failed jobs with exponential backoff, giving transient failures a chance to self-heal.

In this step, get ready to:
  • Simulate random job failures and watch them disappear.
  • Add retry logic with exponential backoff so failed jobs re-enqueue automatically.
  • Verify retries through terminal logs and the job status endpoint.
Simulate random job failures

Before you can build retry logic, you need to see what happens when jobs fail. You will add a 40% random failure rate to the worker, so roughly 4 out of every 10 jobs crash during processing.

  • In main.py, add import random after the import json line at the top of the file.
  • Below the REDIS_URL line, add this constant:
FAILURE_RATE = 0.4
  • In the worker_loop function, find this block of code (everything after job_id = job["id"] down to and including the print line):
            await redis_client.hset(
                f"job:{job_id}",
                mapping={"status": "processing", "worker": name},
            )

            await asyncio.sleep(0.5)

            await redis_client.hset(
                f"job:{job_id}",
                mapping={
                    "status": "completed",
                    "worker": name,
                    "completed_at": str(time.time()),
                },
            )
            print(f"[{name}] Processed job {job_id}: completed")
  • Replace that entire block with the following code:
            # Read the attempt counter from the job payload
            attempt = job.get("attempt", 1)

            await redis_client.hset(
                f"job:{job_id}",
                mapping={"status": "processing", "worker": name, "attempt": attempt},
            )

            try:
                # Randomly fail ~40% of jobs to simulate real failures
                if random.random() < FAILURE_RATE:
                    raise Exception("Simulated processing failure")

                await asyncio.sleep(0.5)

                await redis_client.hset(
                    f"job:{job_id}",
                    mapping={
                        "status": "completed",
                        "worker": name,
                        "attempt": attempt,
                        "completed_at": str(time.time()),
                    },
                )
                print(f"[{name}] Processed job {job_id}: completed (attempt {attempt})")

            except Exception as e:
                print(f"[{name}] Job {job_id} failed: {e}")

What does this code do?

  • attempt = job.get("attempt", 1) reads the retry counter from the job payload, defaulting to 1 for brand-new jobs.
  • The try block wraps all processing logic so you can catch failures.
  • random.random() < FAILURE_RATE simulates a 40% chance of failure by raising an exception.
  • The except block currently just logs the error. The failed job disappears from the queue forever.
  • Save main.py. The --reload flag auto-restarts the app.

Before you submit any jobs, predict what will happen. If roughly 40% of jobs fail and the except block only logs the error, what status will a failed job be stuck at?

  • Return to the Swagger UI at http://127.0.0.1:8000/docs.
  • Submit 5 jobs by clicking the POST /jobs endpoint, then Try it out, then Execute. Repeat this 5 times.
  • Switch back to the terminal where uvicorn is running and check the logs.

You will see some jobs complete normally and others just log a failure message and vanish. Copy a failed job's ID from the logs, then check its status using the GET /jobs/{job_id} endpoint. The status is stuck at processing. No retry, no record of the error, no way to recover the job.

That is exactly the problem. Failed jobs silently disappear from the queue with no chance to recover. The next chunk fixes this.

All 5 jobs completed successfully?

With a 40% failure rate, it is possible (though unlikely) for all 5 to succeed. Submit another 5 jobs. With 10 total, you should see at least a few failures.

If you see no failures at all, make sure you saved main.py and that the terminal shows uvicorn reloading.

Help me figure out why my jobs are not failing randomly.

Add retry logic with exponential backoff

Now that you have seen jobs vanish on failure, you will replace the bare except block with real retry logic. When a job fails, the worker will re-enqueue it with an increasing delay between attempts.

  • In main.py, add this constant below FAILURE_RATE = 0.4:
MAX_RETRIES = 3
  • In the worker_loop function, find this except block:
            except Exception as e:
                print(f"[{name}] Job {job_id} failed: {e}")
  • Replace those two lines with the following retry logic:
            except Exception as e:
                error_msg = str(e)
                if attempt < MAX_RETRIES:
                    next_attempt = attempt + 1
                    # Exponential backoff: 2s, 4s, 8s, capped at 16s
                    backoff = min(2 ** attempt, 16)
                    print(
                        f"[{name}] Job {job_id} failed (attempt {attempt}/{MAX_RETRIES}), retrying in {backoff}s..."
                    )
                    await redis_client.hset(
                        f"job:{job_id}",
                        mapping={
                            "status": "retrying",
                            "attempt": attempt,
                            "last_error": error_msg,
                        },
                    )
                    await asyncio.sleep(backoff)
                    job["attempt"] = next_attempt
                    await redis_client.lpush("queue:jobs", json.dumps(job))
                else:
                    print(
                        f"[{name}] Job {job_id} permanently failed after {MAX_RETRIES} attempts"
                    )
                    await redis_client.hset(
                        f"job:{job_id}",
                        mapping={
                            "status": "failed",
                            "attempt": attempt,
                            "last_error": error_msg,
                        },
                    )

What does this retry logic do?

  • if attempt < MAX_RETRIES checks whether the job has retries left. With MAX_RETRIES = 3, a job gets 3 chances before giving up.
  • min(2 ** attempt, 16) calculates the delay. Attempt 1 waits 2 seconds, attempt 2 waits 4 seconds, attempt 3 waits 8 seconds. The cap prevents absurdly long waits.
  • After the backoff delay, the worker pushes the job back onto the queue with LPUSH and an incremented attempt counter. The next worker pick-up starts a fresh processing attempt.
  • If all retries are exhausted, the job status is set to failed and the error is recorded. The job stays visible but stops retrying.

What is exponential backoff?

Exponential backoff doubles the wait time between each retry. Instead of hammering a failing service every second, the system waits 2 seconds, then 4, then 8.

This gives transient issues (a momentary network hiccup, a brief resource spike) time to resolve before the next attempt. Production systems also add random jitter to prevent many retries from hitting at the exact same moment.

  • Save main.py and wait for uvicorn to reload.

Before testing, you need to update the job creation endpoint so new jobs include an initial attempt counter. Otherwise, the retry logic cannot track which attempt each job is on.

Seeing a syntax error in the terminal?

Make sure the indentation of the except line aligns exactly with the try line above it. Python is sensitive to indentation. The if and else blocks inside the except should be indented one level deeper.

Help me fix a syntax error in my retry logic.

Update job creation to track attempts

New jobs need an attempt field in their payload so the retry logic can track how many times each job has been tried.

  • In the create_job function, find the job dictionary. It currently looks like this:
    job = {
        "id": job_id,
        "payload": payload or {},
        "created_at": str(time.time()),
    }
  • Add "attempt": 1, to the dictionary so it looks like this:
    job = {
        "id": job_id,
        "payload": payload or {},
        "attempt": 1,
        "created_at": str(time.time()),
    }
  • Right below that, find the hset call:
    await redis_client.hset(
        f"job:{job_id}", mapping={"status": "queued"}
    )
  • Add "attempt": 1 to the mapping so it looks like this:
    await redis_client.hset(
        f"job:{job_id}", mapping={"status": "queued", "attempt": 1}
    )
  • Save main.py.

Before you test, predict what will happen. You submit 10 jobs with a 40% failure rate and 3 retries. Do you think any jobs will permanently fail, or will retries save all of them?

  • Return to the Swagger UI at http://127.0.0.1:8000/docs.
  • Submit 10 jobs by clicking POST /jobs, then Try it out, then Execute. Repeat this 10 times.
  • Switch back to the terminal and watch the logs.

You should see retry messages like [Worker-1] Job <id> failed (attempt 1/3), retrying in 2s... appearing in the terminal. Some jobs succeed on the first try, others need a second or third attempt, and a few may permanently fail after 3 tries.

Not seeing retry messages?

Check that MAX_RETRIES = 3 is defined at the top of main.py (not inside a function). Also confirm the except block references MAX_RETRIES and not a different variable name.

Help me debug why my retry logic is not working.

Verify retries through the status endpoint

The GET /jobs/{job_id} endpoint already returns all fields stored in the job's Redis hash. Now that the worker stores attempt, last_error, and richer statuses, the endpoint surfaces all of that automatically.

  • In the Swagger UI, expand the GET /jobs/{job_id} endpoint.
  • Click Try it out.
  • Paste a job ID from the terminal logs into the job_id field. Pick one that had retries.
  • Click Execute.

The response shows the job's current status (one of queued, processing, retrying, completed, or failed), the attempt count, and the last_error message if it failed.

Why is this useful?

In production, operators need visibility into job health. Knowing that a job is on attempt 2 of 3 with a specific error message tells you whether the system is self-healing or whether you need to investigate a deeper issue.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and that uvicorn reloaded without errors.

ⓧ I'd like to double check the full code

The full main.py is over 30 lines, so scroll through your file and confirm each section is present in this order:

  • Imports: asyncio, json, random, time, uuid, redis.asyncio, FastAPI, asynccontextmanager.
  • Constants: REDIS_URL, MAX_RETRIES = 3, FAILURE_RATE = 0.4.
  • Global redis_client set to None.
  • worker_loop function with attempt tracking, random failure, try/except with exponential backoff retry, and CancelledError handling.
  • lifespan function that creates the Redis client and a single worker task.
  • FastAPI app with POST /jobs (including "attempt": 1 in the job dict and hset), GET /jobs/{job_id}, and GET /queue/stats.

Your job queue now handles failure like a production system. Failed jobs get a second and third chance with increasing wait times. But what happens to the jobs that use up all their retries? Right now they just sit with a failed status and no way to recover them. Next up, you will build a Dead Letter Queue to catch and replay permanently failed jobs.

Implement the Dead Letter Queue

Your retry logic from the last step is a big upgrade. Jobs that hit a transient failure now get a second (and third) chance instead of vanishing immediately.

But what happens to jobs that exhaust all three retries? Right now, they log a final error and disappear. In production, you need a place to inspect those permanently failed jobs, figure out what went wrong, and re-run them once the root cause is fixed. That place is a Dead Letter Queue (DLQ). In this step, you will build one.

In this step, get ready to:
  • Route jobs that exhaust all retries into a Dead Letter Queue in Redis.
  • Add API endpoints to inspect and replay dead-lettered jobs.
  • Update the stats endpoint to include DLQ depth and test the full flow.
Route exhausted jobs to the Dead Letter Queue

When a job exhausts all its retries, you want two things to happen: the job's status should update to dead_lettered, and the full job payload should land in a separate Redis list so you can inspect it later.

  • In main.py, scroll to the else branch inside the worker_loop function's exception handler.
  • Find this block of code:
                else:
                    print(
                        f"[{name}] Job {job_id} permanently failed after {MAX_RETRIES} attempts"
                    )
                    await redis_client.hset(
                        f"job:{job_id}",
                        mapping={
                            "status": "failed",
                            "attempt": attempt,
                            "last_error": error_msg,
                        },
                    )
  • Replace the entire block above with the following code:
                else:
                    # Job exhausted all retries, move it to the DLQ
                    print(
                        f"[{name}] Job {job_id} permanently failed after {MAX_RETRIES} attempts, moving to DLQ"
                    )
                    await redis_client.hset(
                        f"job:{job_id}",
                        mapping={
                            "status": "dead_lettered",
                            "attempt": attempt,
                            "last_error": error_msg,
                        },
                    )
                    # Attach error details and push to the DLQ list
                    job["last_error"] = error_msg
                    job["dead_lettered_at"] = str(time.time())
                    await redis_client.lpush("dlq:jobs", json.dumps(job))

What changed here?

Three things are different from the old version:

  • The status changed from failed to dead_lettered so you can distinguish permanently failed jobs from ones still retrying.
  • The job payload now gets the error message and a timestamp attached to it before being pushed onto the dlq:jobs Redis list.
  • LPUSH adds the job to a completely separate list from the main queue. This is the Dead Letter Queue.
  • Save main.py.
  • Check your terminal. You should see uvicorn detect the change and reload the app.

Seeing a syntax error in the terminal?

  • Make sure the new lines sit inside the else block at the same indentation level as the print() and hset() calls above them.
  • Confirm you deleted the entire old else block before pasting the new one. Leftover lines from the old version cause indentation mismatches.

Still stuck? Ask for help.

Add endpoints to inspect and replay dead-lettered jobs

Jobs are flowing into the DLQ now, but you have no way to see them. You need two endpoints: one to list everything in the DLQ, and one to replay a specific job by pulling it out and re-enqueuing it.

  • Add the following two endpoints at the bottom of main.py, after the existing queue_stats function:
@app.get("/dlq")
async def list_dlq():
    # Fetch every job sitting in the Dead Letter Queue
    raw_jobs = await redis_client.lrange("dlq:jobs", 0, -1)
    return [json.loads(j) for j in raw_jobs]


@app.post("/dlq/{job_id}/replay")
async def replay_dlq_job(job_id: str):
    # Scan the DLQ for the matching job
    raw_jobs = await redis_client.lrange("dlq:jobs", 0, -1)
    for i, raw in enumerate(raw_jobs):
        job = json.loads(raw)
        if job["id"] == job_id:
            # Remove from DLQ, reset attempt counter, re-enqueue
            await redis_client.lrem("dlq:jobs", 1, raw)
            job["attempt"] = 1
            job.pop("last_error", None)
            job.pop("dead_lettered_at", None)
            await redis_client.hset(
                f"job:{job_id}", mapping={"status": "queued", "attempt": 1}
            )
            await redis_client.lpush("queue:jobs", json.dumps(job))
            return {"status": "replayed", "job_id": job_id}
    return {"error": "Job not found in DLQ"}

What do these endpoints do?

  • GET /dlq uses LRANGE to read every entry in the dlq:jobs list without removing them. You get a snapshot of everything that permanently failed.
  • POST /dlq/{job_id}/replay finds a specific job, removes it from the DLQ with LREM, resets its attempt counter back to 1, and pushes it onto the main queue for a fresh round of processing.

Why scan instead of direct lookup?

Redis lists do not support direct key-based lookups. The replay endpoint scans the DLQ with LRANGE to find the right job. This O(n) approach works well for a learning project. Production systems often use a Redis sorted set for faster access.

  • Save main.py.
  • Return to the Swagger docs at http://127.0.0.1:8000/docs and refresh the page.
  • You should see the new GET /dlq and POST /dlq/{job_id}/replay endpoints listed.

Now test the full DLQ flow. You need to submit enough jobs that at least one exhausts all three retries and lands in the DLQ.

  • Expand the POST /jobs endpoint in the Swagger docs.
  • Click Try it out, then click Execute about 15 times to submit jobs. Click quickly to build up a stream of work.
  • Watch your terminal. You should see a mix of successful completions, retries, and eventually a message like [Worker-1] Job <id> permanently failed after 3 attempts, moving to DLQ.
  • Wait about 30 seconds for retries and backoff delays to finish.
  • Expand GET /dlq in the Swagger docs, click Try it out, then click Execute.
  • You should see a JSON array containing the dead-lettered jobs, each with their last_error and dead_lettered_at fields.

GET /dlq returning an empty array?

  • With a 40% failure rate and 3 retries, only about 6% of jobs land in the DLQ. If you submitted fewer than 15 jobs, try submitting more and waiting another 30 seconds for backoff delays to complete.
  • Check the terminal for any moving to DLQ messages. If you see none, submit another batch of 15 jobs.
  • Confirm that your else block uses dlq:jobs as the Redis key (not queue:dlq or another name). The list_dlq endpoint reads from the same key.

Need a hand? Ask for help.

Update stats and test the replay flow

Your stats endpoint should also reflect how many jobs are sitting in the DLQ. A quick addition to queue_stats makes that visible.

  • In main.py, find the queue_stats function. It currently looks like this:
@app.get("/queue/stats")
async def queue_stats():
    queue_depth = await redis_client.llen("queue:jobs")
    return {
        "queue_depth": queue_depth,
    }
  • Replace the entire queue_stats function with this updated version:
@app.get("/queue/stats")
async def queue_stats():
    queue_depth = await redis_client.llen("queue:jobs")
    # Include the DLQ depth so you can monitor failed jobs
    dlq_depth = await redis_client.llen("dlq:jobs")
    return {
        "queue_depth": queue_depth,
        "dlq_depth": dlq_depth,
    }
  • Save main.py.

Before you check the stats, think about what you expect. You submitted about 15 jobs earlier. Some completed, some retried and then completed, and a few should have landed in the DLQ. What do you think dlq_depth will show?

  • In the Swagger docs, expand GET /queue/stats, click Try it out, then click Execute.
  • You should see a response with both queue_depth (likely 0 if all jobs finished processing) and dlq_depth showing the number of dead-lettered jobs.

Now for the real payoff. You are going to replay a dead-lettered job and watch it re-enter the main queue.

  • Expand GET /dlq and click Execute. Copy the id value from the first job in the response.
  • Expand POST /dlq/{job_id}/replay, click Try it out, and paste the job ID into the job_id field.
  • Click Execute.

Do you think the replayed job will succeed or fail again? Remember, the 40% failure rate still applies, but the job gets a fresh set of 3 attempts.

Check your terminal. You should see the worker pick up the replayed job and process it. The response from the replay endpoint shows {"status": "replayed", "job_id": "..."} confirming the job left the DLQ and re-entered the main queue.

Replay returning "Job not found in DLQ"?

  • Double check the job ID you pasted. It should be a full UUID like a1b2c3d4-e5f6-7890-abcd-ef1234567890 with no extra spaces or quotes around it.
  • Make sure you copied the id from the GET /dlq response, not from GET /jobs/{job_id}. The job must actually be in the DLQ to replay it.

Still having trouble? Ask for help.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py with all your changes.

ⓧ I'd like to double check the full code

Your main.py should now contain these sections, in order. Scroll up through the step and confirm each one is present:

  • Imports (asyncio, json, random, time, uuid, aioredis, FastAPI, asynccontextmanager) and constants at the top, unchanged from Step 3.
  • worker_loop function with the updated else branch that sets status to dead_lettered and pushes to dlq:jobs.
  • lifespan function and app declaration, unchanged.
  • Existing endpoints: POST /jobs, GET /jobs/{job_id}, unchanged.
  • Updated GET /queue/stats now including dlq_depth.
  • New endpoints: GET /dlq and POST /dlq/{job_id}/replay at the bottom of the file.

Your job queue now handles the full lifecycle: jobs get queued, processed, retried on failure, and dead-lettered when they exhaust all attempts. You can inspect the damage and replay jobs with a single API call. The one remaining problem is speed. Right now, a single worker processes everything. In the next step, you will build an auto-scaling worker pool that spins up more workers when the queue fills up.

Build an Auto-Scaling Worker Pool

In the last step, you built a Dead Letter Queue so permanently failed jobs have a safe place to land and can be replayed later. Your queue system now handles the full lifecycle of a job, from submission to completion, retry, or dead letter.

There is still a bottleneck. Your system has a single worker processing one job at a time. If you flood the queue with 50 jobs, they trickle through one-by-one and the queue backs up. In this step, you will build an auto-scaling worker pool that dynamically spins up more asyncio worker tasks when load spikes and scales back down when the queue drains.

In this step, get ready to:
  • Create a WorkerPool class that manages multiple named asyncio worker tasks.
  • Add an auto-scaler that adjusts worker count based on queue depth.
  • Add a batch endpoint and watch the pool scale up and down in real time.
Create the WorkerPool class

Right now, your app creates a single asyncio.create_task(worker_loop("Worker-1")) in the lifespan. To scale dynamically, you need a manager that can create and cancel worker tasks on demand.

  • In main.py, find the constants at the top of the file. Locate this block:
MAX_RETRIES = 3
FAILURE_RATE = 0.4

redis_client: aioredis.Redis = None
  • Replace it with the following, adding four new scaling constants and a worker_pool global:
MAX_RETRIES = 3
MIN_WORKERS = 1
MAX_WORKERS = 10
SCALE_INTERVAL = 2
JOBS_PER_WORKER = 5
FAILURE_RATE = 0.4

redis_client: aioredis.Redis = None
worker_pool: "WorkerPool" = None

What do these constants control?

  • MIN_WORKERS and MAX_WORKERS set the floor and ceiling for how many concurrent worker tasks can exist.
  • SCALE_INTERVAL is how often (in seconds) the auto-scaler checks queue depth.
  • JOBS_PER_WORKER controls the ratio: one worker for every 5 queued jobs.

Now you will build the class that worker_pool will hold.

  • In main.py, add the WorkerPool class directly below the worker_pool: "WorkerPool" = None line you just added. Copy the following code:
class WorkerPool:
    def __init__(self):
        self.workers: dict[str, asyncio.Task] = {}
        self.active_count = 0

    async def scale_to(self, target: int):
        # Clamp target between min and max
        target = max(MIN_WORKERS, min(target, MAX_WORKERS))
        current = len(self.workers)
        if target == current:
            return
        if target > current:
            # Spin up new worker tasks
            for i in range(current + 1, target + 1):
                name = f"Worker-{i}"
                if name not in self.workers:
                    task = asyncio.create_task(worker_loop(name))
                    self.workers[name] = task
        elif target < current:
            # Cancel excess workers, highest-numbered first
            workers_to_remove = sorted(
                self.workers.keys(), reverse=True
            )[: current - target]
            for name in workers_to_remove:
                self.workers[name].cancel()
                del self.workers[name]
        print(f"[AutoScaler] Scaled workers {current} -> {len(self.workers)}")

    def count(self) -> int:
        return len(self.workers)

What does WorkerPool do?

WorkerPool manages a dictionary of named asyncio tasks. Each key is a worker name like Worker-1 or Worker-5, and each value is a running asyncio task.

The scale_to() method compares the current worker count to the target. If more are needed, it creates new tasks by calling asyncio.create_task(worker_loop(name)). If fewer are needed, it cancels the highest-numbered workers first.

Before you update the lifespan, there is one small refactor to make inside worker_loop.

  • At the top of main.py, find the imports section. Locate this line:
import asyncio
import json
import random
import time
import uuid
  • Remove import random from this block, so it looks like this:
import asyncio
import json
import time
import uuid
  • Now scroll down to the worker_loop function. Inside the inner try: block, find this line:
            try:
                if random.random() < FAILURE_RATE:
  • Add import random as the first line inside the try: block, so it looks like this:
            try:
                import random

                if random.random() < FAILURE_RATE:

Why move the import?

Moving import random inside the function keeps each worker self-contained. Python caches imports, so there is no performance cost. It is a common pattern for imports used in only one function.

Now update the lifespan to use your new WorkerPool instead of a single task.

  • In main.py, find the existing lifespan function. Replace the entire function with this updated version:
@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_client, worker_pool
    redis_client = aioredis.from_url(REDIS_URL, decode_responses=True)
    await redis_client.ping()
    print("[Startup] Connected to Redis")

    # Initialize pool and start with 1 worker
    worker_pool = WorkerPool()
    await worker_pool.scale_to(1)

    yield

    # Cancel all workers on shutdown
    for task in worker_pool.workers.values():
        task.cancel()
    await redis_client.aclose()
    print("[Shutdown] Cleaned up")

What changed in the lifespan?

  • worker_pool is added to the global declaration so the rest of the app can access it.
  • Instead of creating a single worker_task, the lifespan now creates a WorkerPool and calls scale_to(1) to start with one worker.
  • On shutdown, the for loop cancels every worker task in the pool, not just one.
  • Save main.py.
  • Switch back to your terminal. If uvicorn is running with --reload, it should restart automatically. Look for [Startup] Connected to Redis and [AutoScaler] Scaled workers 0 -> 1 in the terminal output.
  • Open the Swagger UI at http://127.0.0.1:8000/docs and submit a test job using the POST /jobs endpoint. You should see [Worker-1] Processed job <id>: completed in the terminal.

Seeing a NameError for WorkerPool?

  • Make sure the WorkerPool class is defined above the worker_loop function. Python reads the file top-to-bottom, so scale_to() calls worker_loop at runtime, but the class itself must exist before lifespan references it.
  • Check that you removed import random from the top of the file AND added it inside the try: block in worker_loop. Missing both sides of that move causes a NameError on random.

Help me debug my WorkerPool setup

Add the auto-scaler coroutine

The WorkerPool can scale, but nothing tells it when to. The auto-scaler is a background coroutine that polls queue depth every 2 seconds and adjusts the worker count based on how many jobs are waiting.

  • In main.py, add the auto_scaler coroutine directly below the worker_loop function and above the lifespan function. Copy the following code:
async def auto_scaler():
    global redis_client, worker_pool
    while True:
        await asyncio.sleep(SCALE_INTERVAL)
        # Check how many jobs are waiting
        queue_depth = await redis_client.llen("queue:jobs")
        # Calculate desired workers: 1 per 5 queued jobs
        desired = max(queue_depth // JOBS_PER_WORKER, MIN_WORKERS)
        desired = min(desired, MAX_WORKERS)
        current = worker_pool.count()
        if desired != current:
            print(
                f"[AutoScaler] Queue depth: {queue_depth}, scaling workers {current} -> {desired}"
            )
            await worker_pool.scale_to(desired)

How does the scaling formula work?

The formula min(max(queue_depth // JOBS_PER_WORKER, MIN_WORKERS), MAX_WORKERS) gives 1 worker for 0 to 4 queued jobs, 2 workers for 5 to 9, and so on up to 10 workers at 50 or more jobs.

The auto-scaler only calls scale_to() when the desired count differs from the current count. This avoids unnecessary log noise and task churn when the queue is stable.

Now wire the auto-scaler into the lifespan so it runs alongside the workers.

  • In the lifespan function, find this line:
    await worker_pool.scale_to(1)

    yield
  • Add the auto-scaler task between scale_to(1) and yield, so the block looks like this:
    await worker_pool.scale_to(1)
    scaler_task = asyncio.create_task(auto_scaler())

    yield
  • Still in the lifespan function, find the shutdown section below yield. It currently starts with:
    yield

    # Cancel all workers on shutdown
    for task in worker_pool.workers.values():
  • Add scaler_task.cancel() between yield and the for loop, so it looks like this:
    yield

    scaler_task.cancel()
    for task in worker_pool.workers.values():
        task.cancel()
    await redis_client.aclose()
    print("[Shutdown] Cleaned up")
  • Save main.py. The app should auto-reload.
  • Check the terminal for [Startup] Connected to Redis and [AutoScaler] Scaled workers 0 -> 1. The auto-scaler is now running in the background alongside your worker.

App crashed on reload?

  • Check that auto_scaler is defined above the lifespan function. The lifespan calls asyncio.create_task(auto_scaler()), so the function must already exist at that point in the file.
  • Make sure you removed the inline comment # Cancel all workers on shutdown only if you replaced the full shutdown block. The scaler_task.cancel() line must come before the for loop.

Help me debug my auto-scaler setup

Add the batch endpoint and watch it scale

To properly test the auto-scaler, you need a way to flood the queue with many jobs at once. Clicking the POST /jobs button 50 times is not practical. A batch endpoint solves this.

  • In main.py, add the /jobs/batch endpoint directly below the get_job_status function. Copy the following code:
@app.post("/jobs/batch")
async def create_batch(count: int = 10):
    job_ids = []
    # Use a Redis pipeline to batch all writes
    pipe = redis_client.pipeline()
    for _ in range(count):
        job_id = str(uuid.uuid4())
        job = {
            "id": job_id,
            "payload": {},
            "attempt": 1,
            "created_at": str(time.time()),
        }
        pipe.hset(f"job:{job_id}", mapping={"status": "queued", "attempt": 1})
        pipe.lpush("queue:jobs", json.dumps(job))
        job_ids.append(job_id)
    await pipe.execute()
    return {"jobs_created": count, "job_ids": job_ids}

Why use a Redis pipeline?

Without a pipeline, each HSET and LPUSH would be a separate round-trip to Redis. A pipeline batches all the commands into a single network call, so 50 jobs get enqueued in one shot instead of 100 separate requests.

Finally, update the stats endpoint so you can see how many workers are active.

  • In main.py, find the queue_stats function. It currently returns:
    return {
        "queue_depth": queue_depth,
        "dlq_depth": dlq_depth,
    }
  • Add an active_workers field, so the return statement looks like this:
    return {
        "queue_depth": queue_depth,
        "dlq_depth": dlq_depth,
        "active_workers": worker_pool.count() if worker_pool else 0,
    }
  • Save main.py and wait for the app to reload.

Before you submit the batch, think about what you expect to see. You are about to enqueue 50 jobs. The auto-scaler checks queue depth every 2 seconds. Do you think it will spin up all 10 workers at once, or scale up gradually?

  • In the Swagger UI at http://127.0.0.1:8000/docs, expand the POST /jobs/batch endpoint.
  • Click Try it out.
  • Set the count parameter to 50.
  • Click Execute and immediately switch to your terminal.

Watch the terminal output. You should see the auto-scaler react within a few seconds.

What should I see?

The auto-scaler detects the queue depth spike and scales up. You should see log lines like:

  • [AutoScaler] Queue depth: 45, scaling workers 1 -> 5 as the scaler detects the burst.
  • Multiple [Worker-1], [Worker-2], [Worker-3] log lines processing jobs in parallel.
  • As the queue drains, scale-down lines like [AutoScaler] Queue depth: 2, scaling workers 5 -> 1.

The exact numbers depend on timing. Some jobs fail and retry, so the queue depth fluctuates. The key is seeing the worker count go up under load and back down when it drains.

  • Once the queue drains, hit the GET /queue/stats endpoint in the Swagger UI. You should see active_workers back at 1 and queue_depth at or near 0.

Not seeing any scaling logs?

  • Make sure you set the count parameter to 50 in the Swagger UI. The default is 10, which gives only 2 workers. With a single worker already running and fast processing, you might not see a visible scale event with 10 jobs.
  • Check that scaler_task = asyncio.create_task(auto_scaler()) appears in your lifespan function after scale_to(1). If that line is missing, the auto-scaler never starts.
  • If the app seems slow to react, remember the auto-scaler checks every 2 seconds (SCALE_INTERVAL). Wait a few seconds after submitting the batch.

Help me debug why my auto-scaler is not scaling workers

✔️ Awesome, I've got everything!

Great. Double check that you have saved main.py.

ⓧ I'd like to double check the full code

Your main.py is now quite long. Scroll through it and confirm each section is present in this order:

  • Imports: asyncio, json, time, uuid, redis.asyncio, FastAPI, asynccontextmanager. No import random at the top.
  • Constants: REDIS_URL, MAX_RETRIES, MIN_WORKERS, MAX_WORKERS, SCALE_INTERVAL, JOBS_PER_WORKER, FAILURE_RATE. Then globals redis_client and worker_pool.
  • WorkerPool class: __init__, scale_to(), count().
  • worker_loop function: import random inside the inner try: block.
  • auto_scaler function: polls LLEN every SCALE_INTERVAL seconds.
  • lifespan function: creates WorkerPool(), calls scale_to(1), creates scaler_task. On shutdown, cancels scaler_task and all workers.
  • Endpoints: POST /jobs, GET /jobs/{job_id}, POST /jobs/batch, GET /queue/stats (with active_workers), GET /dlq, POST /dlq/{job_id}/replay.

Your job queue now scales itself. When load spikes, the pool grows to chew through the backlog. When the queue drains, it shrinks back to a single worker. Next up, you will take on the Secret Mission and add priority-based routing so urgent jobs jump the line.

Secret mission

Add Job Priorities

Right now, every job in your queue is treated equally. A critical payment notification waits behind a batch of thumbnail resizes. Real job queues route urgent work first. Can you add high, medium, and low priority levels so important jobs always jump the line?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • A Docker container named redis-dev running Redis on port 6379. This container uses minimal resources but persists until you remove it.
  • Your project folder on the Desktop, containing main.py and a Python virtual environment with FastAPI, uvicorn, and redis-py installed.

Keep everything running

No action needed. Choose this if you are still actively building or want to keep experimenting with the job queue.

  • Your redis-dev Docker container stays running and uses minimal memory.
  • Your project folder, virtual environment, and all installed packages remain on your Desktop.
  • To resume work, activate your virtual environment and run uvicorn main:app --reload from your project folder.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your project files and Docker container data stay intact so you can pick up where you left off.

  • If uvicorn is still running in your terminal, press Ctrl+C to stop it.
  • Stop the Redis container by running:
docker stop redis-dev
  • When you are ready to resume, restart the container by running:
docker start redis-dev
  • Then activate your virtual environment and start the app again with uvicorn main:app --reload.

Delete - I don't want to use this again

Remove the Docker container and your project files to start completely fresh.

  • If uvicorn is still running in your terminal, press Ctrl+C to stop it.
  • Stop and remove the Redis container by running these commands:
docker stop redis-dev
docker rm redis-dev
  • Delete your project folder from the Desktop. On macOS or Linux, navigate to your Desktop and remove it by running:
cd ~/Desktop
rm -rf [[PROJECT_FOLDER="your project folder name"]]
  • On Windows, navigate to your Desktop and remove it by running:
cd ~\Desktop
rmdir /s /q [[PROJECT_FOLDER="your project folder name"]]
  • This removes the virtual environment and all installed packages along with your code. Python, VS Code, and Docker Desktop remain installed on your machine.

Nice Work!

Nice Work!

Nice work! You just built a fully working distributed job queue system powered by FastAPI and Redis. You can submit jobs through a REST API, watch them flow through a queue, retry on failure, land in a Dead Letter Queue when they give up, and auto-scale workers to keep up with demand.

You've learned how to:

  • Build a FastAPI REST API backed by a Redis queue, with an async worker that pops jobs using BRPOP and processes them in the background.
  • Implement retry logic with exponential backoff that automatically re-enqueues failed jobs up to 3 times, and a Dead Letter Queue for jobs that exhaust retries, with endpoints to inspect and replay them.
  • Create an auto-scaling worker pool that dynamically adjusts from 1 to 10 concurrent asyncio tasks based on queue depth, reacting to load in real time.
  • Secret Mission: Add priority-based routing with three queue levels so urgent jobs always process first.

Ready to quiz yourself?