Build a URL Shortener API with Redis

Build a production-hardened URL shortener backed by Redis with rate limiting.

Introduction

30 Second Summary

Every time you share a link on social media or in a message, there's a good chance a URL shortener turned a messy address into something clean and trackable. Behind that simple redirect is a system handling thousands of requests per second, blocking abuse, and counting every click.

In this project, you will build a production-hardened URL shortener API from scratch, using Redis as your data store, rate limiter, and message queue all at once.

What You'll Build

You can POST any long URL, get back a short code, follow it to get redirected, and watch a background worker tally click counts in real time as requests flow through your rate limiter.

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

  • A working URL shortener where you paste a long link, get back a short code, and watch it redirect in the browser.
  • A custom sliding window rate limiter that visibly blocks you with a 429 response after 10 rapid-fire requests, proving your API can defend itself from abuse.
  • A background analytics worker running in a separate terminal that prints each click event as it processes them from a Redis queue in real time.
  • Secret Mission: Add a Redis caching layer with TTL for popular redirects and build a stats endpoint exposing click analytics.

Are there any prerequisites?

You should have basic Python experience and some familiarity with REST APIs. A Django background is more than sufficient. You do not need prior experience with FastAPI or Redis.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters. This is your moment to commit to the goal before any hands-on work begins.

Set Up Your Environment

Every backend project needs its runtime, dependencies, and data layer in place before you can write a single route.

In this step, you'll create a project folder, install FastAPI and redis-py, spin up a Redis server in Docker, and confirm everything works by running a minimal app.

In this step, get ready to:
  • Create a project folder, virtual environment, and install dependencies.
  • Start a Redis server in Docker.
  • Run a minimal FastAPI app and see the Swagger UI.
Create your project directory and install dependencies
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type VS Code and press Enter to open it.
  • Open the integrated terminal by pressing Ctrl+` (backtick key, above Tab) on both macOS and Windows.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir url-shortener
cd url-shortener
  • Confirm you're inside the new folder by running:
pwd

You should see a path ending in url-shortener, like /Users/yourname/Desktop/url-shortener.

  • Create a Python virtual environment and activate it by running these commands:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

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

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay scoped to this project and don't affect other Python projects on your machine.

Don't see (venv) in your prompt?

Make sure you ran the activate command for your operating system. On macOS it's source venv/bin/activate. On Windows it's venv\Scripts\activate.

If you see a permissions error on Windows, try running Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned in PowerShell first, then retry the activate command.

Help me fix this

  • Install the project dependencies by running:
pip install "fastapi[standard]==0.141.1" redis==8.1.0

What does this install?

  • fastapi[standard] installs FastAPI plus Uvicorn (the production server that runs your app).
  • redis installs redis-py, the Python client for talking to a Redis server.

You should see pip download several packages and finish with a line containing Successfully installed.

  • Now open the url-shortener folder in VS Code so the Explorer sidebar shows your project files.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop, select the url-shortener folder, and click Open.

Why open the folder?

Opening the folder (not just a file) makes VS Code's Explorer sidebar show all your project files. It also sets the integrated terminal's working directory to your project root automatically.

  • If VS Code asks whether you trust the authors of the folder, click Yes, I trust the authors.
  • Open the integrated terminal again by pressing Ctrl+` (backtick).
  • Reactivate your virtual environment in the new terminal:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate
  • Create a requirements.txt file to record your dependencies. In the Explorer sidebar, click the New File icon (paper with a +) next to URL-SHORTENER.
  • Name the file requirements.txt and press Enter.
  • Paste the following into the file:
fastapi[standard]==0.141.1
redis==8.1.0

What is requirements.txt?

This file records the exact package versions your project depends on. Anyone who clones your project can run pip install -r requirements.txt to recreate the same environment.

  • Save requirements.txt with Cmd+S (macOS) or Ctrl+S (Windows).
Start Redis via Docker

Your URL shortener needs somewhere to store data. Redis is an in-memory data store that you'll run as a Docker container so it's isolated and easy to tear down later.

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

What does this command do?

  • -d runs the container in the background so your terminal stays free.
  • --name redis gives the container a friendly name you can reference later.
  • -p 6379:6379 maps the container's Redis port to your machine so your Python code can connect.
  • redis:7 pulls the official Redis 7 image from Docker Hub.
  • Verify Redis is running by pinging it:
docker exec redis redis-cli ping

You should see PONG printed in your terminal. This confirms Redis is alive and responding to commands.

Seeing a Docker error or no PONG?

Make sure Docker Desktop is running. Look for the Docker whale icon in your system tray (Windows) or menu bar (macOS). If it's not there, open Docker Desktop and wait for it to finish starting.

If you see a message about the name "redis" already being in use, a previous container exists. Remove it by running docker rm -f redis and then try the docker run command again.

Help me troubleshoot Docker

Create your FastAPI app and run it

With dependencies installed and Redis running, you can now create the application entry point and confirm everything connects.

  • In the Explorer sidebar, click the New File icon next to URL-SHORTENER.
  • Name the file main.py and press Enter.
  • Paste the following code into main.py:
from fastapi import FastAPI

app = FastAPI()

What does this code do?

This creates a FastAPI application object. Right now it has no routes, but it's enough to start the dev server and confirm the auto-generated documentation page loads.

  • Save main.py with Cmd+S (macOS) or Ctrl+S (Windows).

Before you run the server, what do you think you'll see when you visit the app in a browser with no routes defined?

  • Start the FastAPI development server by running:
fastapi dev main.py

Uvicorn starts and prints a line showing your app is running at http://127.0.0.1:8000.

Keep the server running

Leave the fastapi dev process running in this terminal tab. When you need to run other commands in later steps, open a new terminal tab with Ctrl+Shift+` in VS Code.

  • Open your browser and visit http://127.0.0.1:8000/docs.

You should see the Swagger UI page with your app title but no endpoints listed yet. This confirms FastAPI is running and auto-generating documentation for you.

Server not starting or page not loading?

If you see a message about no module named fastapi, your virtual environment is not active. Check that your prompt shows (venv) and reactivate if needed.

If the browser shows a connection refused error, make sure the terminal is still running the server. The fastapi dev command takes over that terminal window. You need it running to keep the app alive.

Help me troubleshoot the server

✔️ Awesome, I've got everything!

Great. Confirm you've saved both main.py and requirements.txt, and that the Swagger UI is visible at http://127.0.0.1:8000/docs.

ⓧ I'd like to double check the full code

Here is what each file should contain at this point.

main.py:

from fastapi import FastAPI

app = FastAPI()

requirements.txt:

fastapi[standard]==0.141.1
redis==8.1.0

Your environment is ready. You have FastAPI serving requests, Redis waiting for data, and a clean project structure to build on. Next up, you'll create the URL shortener endpoints that store and retrieve links from Redis.

Build the URL Shortener

Your environment is locked in. Python, FastAPI, Redis, and Docker are all running and talking to each other.

Now it's time to build the core feature: a URL shortener that stores and retrieves data from Redis. A working shortener gives you visible proof that Redis hashes are storing and retrieving data. It also gives you a real API to harden in the steps that follow.

In this step, get ready to:
  • Connect FastAPI to Redis using a lifespan context manager.
  • Build POST and GET endpoints that shorten and redirect URLs.
  • Add a health check endpoint and verify the full flow end to end.
Connect FastAPI to Redis and define your models

Before any route can store or fetch URLs, your FastAPI app needs a live connection to Redis. A lifespan context manager handles this by opening the connection when the app starts and closing it when the app shuts down.

  • Replace the entire contents of main.py with the following connection setup:
import time
import string
import random
import json
from contextlib import asynccontextmanager

import redis.asyncio as aioredis
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import RedirectResponse, JSONResponse
from pydantic import BaseModel

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

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Open a Redis connection when the app starts
    global redis_client
    redis_client = aioredis.from_url("redis://localhost:6379", decode_responses=True)
    yield
    # Close the connection when the app shuts down
    await redis_client.close()

app = FastAPI(title="URL Shortener", lifespan=lifespan)

What does this code do?

  • redis.asyncio as aioredis gives you an async Redis client that works with FastAPI's async routes.
  • The lifespan function runs once at startup (opening the Redis connection) and once at shutdown (closing it). Everything between yield is where your app serves requests.
  • decode_responses=True means Redis returns Python strings instead of raw bytes.
  • Save main.py.
  • Check your terminal where fastapi dev is running. The server should restart automatically with no errors.

Seeing a connection error on startup?

If you see a traceback mentioning connection refused, your Redis container may have stopped. Run docker start redis in a new terminal to restart it.

Help me fix my Redis connection error

Now add the data models and helper function that your routes will use.

  • Add this code below the app = FastAPI(...) line in main.py:
class URLCreate(BaseModel):
    url: str

class URLResponse(BaseModel):
    short_code: str
    short_url: str
    original_url: str

def generate_short_code(length: int = 6) -> str:
    # Pick random letters and digits to form a short code
    chars = string.ascii_letters + string.digits
    return "".join(random.choices(chars, k=length))

What does this code do?

  • URLCreate is a Pydantic model that validates incoming request bodies. It expects a single url string field.
  • URLResponse defines the shape of data your API sends back: the short code, the full short URL, and the original URL.
  • generate_short_code() creates a random 6-character string from letters and digits. This becomes the short URL's unique identifier.
  • Save main.py and confirm the server restarts cleanly in the terminal with no errors.
Implement the shorten and redirect endpoints

With your models and helper ready, you can now build the two routes that make a URL shortener work: one to create short URLs, and one to follow them.

  • Add the POST /shorten endpoint below generate_short_code in main.py:
@app.post("/shorten", response_model=URLResponse)
async def shorten_url(payload: URLCreate, request: Request):
    short_code = generate_short_code()

    # Store URL data as a Redis hash with fields for the original URL, timestamp, and click counter
    await redis_client.hset(f"url:{short_code}", mapping={
        "original_url": str(payload.url),
        "created_at": str(time.time()),
        "clicks": "0",
    })

    base_url = str(request.base_url).rstrip("/")
    return URLResponse(
        short_code=short_code,
        short_url=f"{base_url}/{short_code}",
        original_url=str(payload.url),
    )

What does this code do?

  • It generates a random short code, then stores the URL in a Redis hash at the key url:{short_code}.
  • A Redis hash is like a mini dictionary. Each hash holds multiple fields (original_url, created_at, clicks) without needing a separate key for each.
  • The response includes the full short URL so the client knows exactly where to redirect.
  • Save main.py.
  • Test the endpoint by running this command in a new terminal:
curl -X POST http://127.0.0.1:8000/shorten -H "Content-Type: application/json" -d '{"url": "https://github.com"}'

You should see a JSON response containing a short_code, short_url, and original_url field.

Getting a 422 or connection error?

  • A 422 error means the request body is malformed. Make sure you include both the -H "Content-Type: application/json" header and the -d flag with valid JSON.
  • If you see a 500 error, check that your Redis container is still running with docker ps.

Help me debug my /shorten endpoint

Now add the redirect route that looks up a short code and sends the user to the original URL.

  • Add the GET /{short_code} endpoint below the shorten_url function in main.py:
@app.get("/{short_code}")
async def redirect_url(short_code: str):
    # Fetch all fields from the Redis hash for this short code
    url_data = await redis_client.hgetall(f"url:{short_code}")

    if not url_data:
        raise HTTPException(status_code=404, detail="Short URL not found")

    # Send the user to the original URL with a 307 redirect
    return RedirectResponse(url=url_data["original_url"], status_code=307)

What does this code do?

  • hgetall retrieves all fields from the Redis hash at url:{short_code} as a Python dictionary.
  • If the hash doesn't exist (the code is invalid), the route returns a 404.
  • A 307 redirect tells the browser to go to the original URL while preserving the HTTP method.
  • Save main.py.
  • Copy the short_code from the JSON response you received earlier.
  • Open http://127.0.0.1:8000/{your_short_code} in your browser (replacing {your_short_code} with the actual code).

Your browser should redirect you to GitHub. That's your shortener working end to end.

Not getting redirected?

  • Make sure you're using the exact short code from the curl response. Codes are case-sensitive.
  • If you see a 404, the code may have been lost if the server restarted after you shortened it. Run the curl command again to generate a fresh code.

Help me fix my redirect endpoint

Add a health check and verify the full flow

A health endpoint lets you quickly confirm your app can reach Redis without needing to shorten a URL first. It's also the standard way load balancers check if a service is alive.

  • Add the GET /health endpoint in main.py. Place it below generate_short_code and above the /shorten route:
@app.get("/health")
async def health_check():
    try:
        await redis_client.ping()
        return {"status": "healthy", "redis": "connected"}
    except Exception:
        raise HTTPException(status_code=503, detail="Redis unavailable")

What does this code do?

It sends a PING command to Redis. If Redis responds, the endpoint returns a success JSON. If the connection is broken, it raises a 503 error so monitoring tools know something is wrong.

  • Save main.py.
  • Verify the health endpoint by running:
curl http://127.0.0.1:8000/health

You should see {"status":"healthy","redis":"connected"} in the response.

Seeing 'Redis unavailable'?

Run docker ps and confirm the redis container is listed as running. If it's stopped, run docker start redis to bring it back.

Help me get my health endpoint working

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and the server is running without errors.

ⓧ I'd like to double check the full code

Your main.py should contain these sections in order. Scroll up and confirm each one is present:

  • Imports (time, string, random, json, asynccontextmanager, redis.asyncio, FastAPI, Pydantic).
  • redis_client global and the lifespan context manager.
  • app = FastAPI(title="URL Shortener", lifespan=lifespan).
  • URLCreate and URLResponse Pydantic models.
  • generate_short_code() helper function.
  • GET /health endpoint.
  • POST /shorten endpoint.
  • GET /{short_code} redirect endpoint.

Time to run the full flow from start to finish. Before you run the next command, predict: what do you think the JSON response will contain?

  • Shorten a URL by running:
curl -X POST http://127.0.0.1:8000/shorten -H "Content-Type: application/json" -d '{"url": "https://github.com"}'

You receive a JSON response with a short_code, a full short_url, and the original_url you submitted.

  • Copy the short_code value from the response and open http://127.0.0.1:8000/{your_short_code} in your browser.

Your browser redirects to GitHub. The full cycle works: store a URL in Redis, generate a short code, and redirect anyone who visits it.

Your URL shortener is alive. You can create short links and follow them back to the original. Next up, you'll lock this API down with input validation, security headers, and CORS.

Add Security Hardening

Your URL shortener works. You can create short codes and follow them to get redirected. But right now, the API has three security gaps that would get flagged by any security scanner.

It accepts any string as a URL (including gibberish like not-a-url), it returns no security headers, and it has no CORS policy. In this step, you will close all three gaps.

In this step, get ready to:
  • Add input validation so only real URLs can be shortened.
  • Add security headers middleware to every response.
  • Configure CORS to restrict cross-origin access.
Expose the security gaps

Before fixing anything, see the problems for yourself. Your server should still be running from the previous step.

  • Try shortening an invalid URL by running this command:
curl -X POST http://127.0.0.1:8000/shorten -H "Content-Type: application/json" -d '{"url": "not-a-url"}'

What do you think will happen? Will the API reject the invalid URL or accept it?

It succeeded. The API happily stored not-a-url as if it were a legitimate web address. That is the first gap.

  • Now check the response headers by running:
curl -I http://127.0.0.1:8000/health

You see basic headers like content-type and date. No X-Content-Type-Options, no X-Frame-Options, no Strict-Transport-Security. That is the second and third gap. Time to fix all three.

Upgrade URL validation with Pydantic HttpUrl

Pydantic's HttpUrl type rejects any string that is not a valid HTTP or HTTPS URL. Swapping it in requires two tiny edits.

  • In main.py, find this import line:
from pydantic import BaseModel
  • Replace it with:
from pydantic import BaseModel, HttpUrl
  • Now find the URLCreate class:
class URLCreate(BaseModel):
    url: str
  • Replace str with HttpUrl so it looks like this:
class URLCreate(BaseModel):
    url: HttpUrl
  • Save main.py. The dev server auto-reloads.
  • Re-run the bad URL test:
curl -X POST http://127.0.0.1:8000/shorten -H "Content-Type: application/json" -d '{"url": "not-a-url"'

You now get a 422 validation error. Pydantic caught the invalid input before your route code ever ran.

What is HttpUrl doing?

Pydantic's HttpUrl type checks that the value starts with http:// or https:// and has a valid domain structure. If validation fails, FastAPI automatically returns a 422 response with details about what went wrong.

Still seeing a 200 response?

Make sure you saved main.py after editing. The dev server only reloads on save. Check the terminal running the server for any import errors.

help me fix my HttpUrl validation

Add security headers middleware

Security headers tell browsers how to behave when loading your API's responses. Without them, browsers allow risky defaults like embedding your pages in iframes or guessing content types.

  • In main.py, find the line app = FastAPI(title="URL Shortener", lifespan=lifespan).
  • Add this middleware function directly below it:
@app.middleware("http")
async def security_headers(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    return response

What does this middleware do?

Every HTTP request passes through this function before reaching your route. After your route returns a response, the middleware attaches four headers:

  • X-Content-Type-Options: nosniff prevents browsers from guessing the MIME type.
  • X-Frame-Options: DENY blocks your pages from being embedded in iframes.
  • Strict-Transport-Security tells browsers to only connect via HTTPS for one year.
  • Content-Security-Policy restricts which resources the page can load.
  • Save main.py and verify the headers appear by running:
curl -I http://127.0.0.1:8000/health

You should now see x-content-type-options: nosniff, x-frame-options: DENY, and strict-transport-security in the output.

Headers not showing up?

Check that the @app.middleware("http") decorator is placed after the app = FastAPI(...) line. The app object must exist before you decorate functions on it.

help me debug my security headers middleware

Add CORS middleware

CORS controls which websites can call your API from a browser. Without a policy, browsers block cross-origin requests entirely. You will allow requests from a local frontend at http://localhost:3000.

  • In main.py, find this import line:
from fastapi import FastAPI, Request, HTTPException
  • Add this new import directly below it:
from fastapi.middleware.cors import CORSMiddleware
  • Now find the closing line of your security headers middleware (return response).
  • Add the CORS configuration below it, leaving one blank line between:
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

What does CORSMiddleware do?

When a browser on http://localhost:3000 tries to call your API, it first sends a preflight OPTIONS request. CORSMiddleware intercepts that request and responds with the allowed origins, methods, and headers. Without this, the browser blocks the actual request.

Setting allow_origins to a specific list (instead of *) means only your frontend can make cross-origin requests. This is safer than allowing every origin.

  • Save main.py.

Before you run the final verification, what headers do you expect to see in the response?

  • Run the header check one more time:
curl -I http://127.0.0.1:8000/health

You should see X-Content-Type-Options: nosniff, X-Frame-Options: DENY, and Strict-Transport-Security in the response.

  • Confirm input validation still works by running:
curl -X POST http://127.0.0.1:8000/shorten -H "Content-Type: application/json" -d '{"url": "not-a-url"}'

You get a 422 validation error. Both security layers are working.

Seeing an import error in the terminal?

Make sure from fastapi.middleware.cors import CORSMiddleware is on its own line after the other fastapi imports. A typo in the module path causes an ImportError and the server fails to restart.

help me fix my CORS middleware import

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and your dev server is running without errors.

ⓧ I'd like to double check the full code

Your main.py should now have all of these sections in order. Scroll through your file and confirm each one is present:

  • Imports: time, string, random, json, asynccontextmanager, redis.asyncio, FastAPI, Request, HTTPException, CORSMiddleware, RedirectResponse, JSONResponse, BaseModel, HttpUrl.
  • The redis_client global and lifespan context manager.
  • app = FastAPI(title="URL Shortener", lifespan=lifespan).
  • The security_headers middleware decorated with @app.middleware("http").
  • The app.add_middleware(CORSMiddleware, ...) block.
  • URLCreate with url: HttpUrl and URLResponse.
  • generate_short_code, /health, POST /shorten, and GET /{short_code} routes.

Your API now validates every URL before storing it, attaches security headers to every response, and has a CORS policy restricting browser access. Next up, you will build a rate limiter that blocks abusive clients using Redis sorted sets.

Add Redis-Backed Rate Limiting

Your URL shortener accepts requests, stores URLs, redirects users, and validates input. But it has no concept of how many requests a single client can make.

An attacker could fire thousands of POST requests and fill your Redis memory with junk URLs. In this step, you will build a sliding window rate limiter using Redis sorted sets that caps each client to 10 requests per minute.

In this step, get ready to:
  • Expose the vulnerability by rapid-firing requests at your unprotected API.
  • Implement a sliding window rate limiter using Redis sorted sets and pipelines.
  • Wire the limiter into FastAPI middleware that blocks abusive clients with a 429 response.
Expose the missing rate limit

Right now, every request succeeds regardless of how fast a client sends them. Let's prove it.

Before you run the command below, what do you think will happen if you fire 20 POST requests in rapid succession?

  • In your terminal (open a new tab if your server is running), blast 20 requests at the /shorten endpoint by running:
for i in $(seq 1 20); do curl -s -o /dev/null -w "%{http_code}\n" -X POST http://127.0.0.1:8000/shorten -H "Content-Type: application/json" -d '{"url": "https://github.com"}'; done

All 20 lines print 200. No throttling, no pushback. An attacker could exhaust your Redis memory in minutes.

What just happened?

You proved that the API has zero protection against abuse. Every request got through and created a new short URL in Redis. The rate limiter you build next will stop this.

Implement the sliding window rate limiter

A sliding window rate limiter tracks requests in a time window that moves forward with each second. Redis sorted sets are perfect for this because each member has a score (the timestamp), making it cheap to remove expired entries and count active ones.

  • In main.py, find the app.add_middleware(CORSMiddleware, ...) block. Add the following function below it (above the class URLCreate line):
async def check_rate_limit(key: str, limit: int, window_seconds: int) -> dict:
    now = time.time()
    window_start = now - window_seconds

    # Use a pipeline to batch four Redis commands into one round trip
    pipe = redis_client.pipeline()
    pipe.zremrangebyscore(key, "-inf", window_start)
    pipe.zadd(key, {f"{now}-{random.random()}": now})
    pipe.zcard(key)
    pipe.expire(key, window_seconds)
    results = await pipe.execute()

    # results[2] is the ZCARD count (third command in the pipeline)
    count = results[2]
    return {
        "allowed": count <= limit,
        "remaining": max(0, limit - count),
        "reset": int(now + window_seconds),
    }

What does this code do?

  • ZREMRANGEBYSCORE removes all entries older than the window (60 seconds ago), cleaning out stale requests.
  • ZADD adds the current request with its timestamp as the score. The random suffix ensures uniqueness even for same-millisecond requests.
  • ZCARD counts how many requests remain in the window after cleanup.
  • EXPIRE sets a TTL on the key so it self-destructs if the client goes quiet.
  • All four commands execute in a single Redis pipeline, meaning one network round trip instead of four.
  • Save main.py.
Add the rate limiting middleware and verify

The function alone does nothing until middleware calls it on every incoming request. This middleware extracts the client's IP, checks the limit, and either blocks with a 429 or lets the request through with rate-limit headers attached.

  • In main.py, add the following middleware directly below the check_rate_limit function you just wrote (still above class URLCreate):
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # Let CORS preflight requests through without counting
    if request.method == "OPTIONS":
        return await call_next(request)

    client_ip = request.client.host
    key = f"ratelimit:{client_ip}"

    result = await check_rate_limit(key, limit=10, window_seconds=60)

    if not result["allowed"]:
        return JSONResponse(
            status_code=429,
            content={"detail": "Rate limit exceeded. Try again later."},
            headers={
                "X-RateLimit-Remaining": str(result["remaining"]),
                "X-RateLimit-Reset": str(result["reset"]),
                "Retry-After": "60",
            },
        )

    response = await call_next(request)
    response.headers["X-RateLimit-Remaining"] = str(result["remaining"])
    response.headers["X-RateLimit-Reset"] = str(result["reset"])
    return response

What does this middleware do?

  • OPTIONS requests pass through uncounted so CORS preflight calls never get rate limited.
  • Each client IP gets its own Redis sorted set key (ratelimit:{ip}).
  • If the client exceeds 10 requests in 60 seconds, the middleware returns a 429 status with X-RateLimit-Remaining, X-RateLimit-Reset, and Retry-After headers.
  • Successful responses also get the rate-limit headers so clients can monitor their remaining quota.
  • Save main.py. Your dev server auto-reloads.

Before you re-run the 20-request loop, what do you expect to see this time? How many requests will succeed before the limiter kicks in?

  • First, flush the old rate-limit data from your previous test by running:
docker exec redis redis-cli KEYS "ratelimit:*" | xargs -r docker exec -i redis redis-cli DEL
  • Now blast 20 requests again by running:
for i in $(seq 1 20); do curl -s -o /dev/null -w "%{http_code}\n" -X POST http://127.0.0.1:8000/shorten -H "Content-Type: application/json" -d '{"url": "https://github.com"}'; done

The first 10 lines print 200. Then the remaining lines print 429. Your rate limiter is working.

  • Confirm the rate-limit headers are present by running:
curl -s -I http://127.0.0.1:8000/health

You should see X-RateLimit-Remaining and X-RateLimit-Reset in the response headers.

Still seeing all 200s?

  • Make sure the rate_limit_middleware is defined after check_rate_limit and before the class definitions. Order matters in FastAPI middleware.
  • Check that your dev server reloaded after saving. Look for the reload message in the terminal running fastapi dev main.py.
  • Confirm Redis is still running with docker exec redis redis-cli ping. You should see PONG.

Why is my rate limiter not blocking requests?

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and that your 20-request loop shows the 200-to-429 transition.

ⓧ I'd like to double check the full code

Your main.py is now over 30 lines, so here is a checklist of the blocks you should have in order. Scroll through your file and confirm each is present:

  • Imports (time, string, random, json, asynccontextmanager, redis.asyncio, FastAPI, Pydantic).
  • redis_client global variable.
  • lifespan context manager connecting to Redis.
  • app = FastAPI(title="URL Shortener", lifespan=lifespan).
  • security_headers middleware.
  • app.add_middleware(CORSMiddleware, ...) block.
  • check_rate_limit async function (pipeline with ZREMRANGEBYSCORE, ZADD, ZCARD, EXPIRE).
  • rate_limit_middleware (extracts client IP, calls check_rate_limit, returns 429 or passes through).
  • URLCreate and URLResponse Pydantic models.
  • generate_short_code helper function.
  • GET /health, POST /shorten, and GET /{short_code} route handlers.

Your API now fights back against abusive clients. Next up, you will decouple click analytics from the redirect path using a Redis queue and a background worker.

Add the Analytics Queue

Your API now rejects bad input, adds security headers, and rate limits abusive clients. The next challenge is tracking how often each short URL gets clicked without slowing down the redirect itself.

Right now, the redirect handler does a hash lookup and returns immediately. If you added click counting directly inside that handler, every user would wait for the analytics write to finish before seeing the redirect. By pushing click events to a Redis list and processing them in a separate background worker, you decouple analytics from the user-facing response path. Redirects stay fast, and a worker tallies clicks at its own pace.

In this step, get ready to:
  • Push click events to a Redis list queue from the redirect handler.
  • Create a background worker that consumes events and updates click counts.
  • Test the full producer/consumer flow end to end.
Push click events to the analytics queue

The producer/consumer pattern works like a to-do list. The redirect handler (producer) drops a note saying "this link was clicked" onto a Redis list. A separate worker process (consumer) picks up notes one by one and does the slow work of updating counters.

  • In main.py, scroll down to the redirect_url function at the bottom of the file.
  • Find this line:
    return RedirectResponse(url=url_data["original_url"], status_code=307)
  • Replace the entire redirect_url function with this updated version that pushes a click event before redirecting:
@app.get("/{short_code}")
async def redirect_url(short_code: str):
    url_data = await redis_client.hgetall(f"url:{short_code}")

    if not url_data:
        raise HTTPException(status_code=404, detail="Short URL not found")

    # Serialize a click event and push it to the analytics queue
    event = json.dumps({
        "short_code": short_code,
        "timestamp": time.time(),
    })
    await redis_client.lpush("analytics:queue", event)

    return RedirectResponse(url=url_data["original_url"], status_code=307)

What does this code do?

  • json.dumps() serializes a dictionary containing the short_code and a timestamp into a JSON string that the worker can parse later.
  • redis_client.lpush("analytics:queue", event) pushes the event onto the head of a Redis list called analytics:queue. This is an O(1) operation, so it adds virtually no latency to the redirect.
  • Save main.py. The dev server auto-reloads.
  • Shorten a URL by running this command in a new terminal tab:
curl -X POST http://127.0.0.1:8000/shorten -H "Content-Type: application/json" -d '{"url": "https://github.com"}
  • Copy the short_code value from the JSON response.
  • Trigger a redirect by running this command (replacing the code with your actual short code):
curl -s -o /dev/null -w "%{http_code}" http://127.0.0.1:8000/[[SHORT_CODE="your short code from the response"]]
  • Verify the event landed in the queue by checking its length:
docker exec redis redis-cli LLEN analytics:queue

You should see 1 (or higher if you triggered multiple redirects). The event is sitting in the queue, waiting for a consumer.

Seeing 0 in the queue?

  • Make sure you saved main.py and the dev server reloaded (check the terminal running fastapi dev for a reload message).
  • Confirm you actually visited a valid short code. If you get a 404, the code doesn't exist in Redis.
  • Check that the lpush line is indented inside the function, AFTER the 404 check.

Why isn't my analytics event appearing in the Redis queue?

Create the analytics worker

The queue has events piling up, but nothing is consuming them yet. You need a separate process that blocks on the queue, waiting for new events, and processes each one by incrementing the click counter in the URL's hash.

  • Create a new file called worker.py in your url-shortener folder (same level as main.py).
  • Add the following code to worker.py:
import asyncio
import json

import redis.asyncio as aioredis


async def process_analytics():
    # Connect to the same Redis instance as the API
    client = aioredis.from_url("redis://localhost:6379", decode_responses=True)
    print("Analytics worker started. Waiting for events...")

    try:
        while True:
            # BRPOP blocks until an event is available (FIFO order)
            result = await client.brpop("analytics:queue", timeout=0)

            if result:
                _, event_data = result
                event = json.loads(event_data)
                short_code = event["short_code"]

                # Increment the click counter on the URL's hash
                await client.hincrby(f"url:{short_code}", "clicks", 1)
                clicks = await client.hget(f"url:{short_code}", "clicks")
                print(f"Click processed: /{short_code} | Total clicks: {clicks}")
    except asyncio.CancelledError:
        pass
    finally:
        await client.close()


if __name__ == "__main__":
    asyncio.run(process_analytics())

What does this code do?

  • client.brpop("analytics:queue", timeout=0) blocks indefinitely until an item appears in the list, then pops it from the tail. Combined with the producer's LPUSH (head), this creates a FIFO queue.
  • client.hincrby(f"url:{short_code}", "clicks", 1) atomically increments the clicks field in the URL's hash without touching any other fields.
  • The asyncio.CancelledError handler lets you stop the worker gracefully with Ctrl+C without leaving dangling connections.

Why LPUSH/BRPOP instead of a library like Celery?

Celery adds a heavy dependency graph for a single background task. Redis lists with LPUSH/BRPOP are the fundamental primitive that ALL message queues are built on. Learning the primitive means you can implement this pattern in any language with any Redis client.

  • Save worker.py.
  • Confirm the file exists by running:
ls

You should see main.py, requirements.txt, and worker.py listed in the output.

✔️ Awesome, I've got everything!

Great. Make sure you've saved both main.py and worker.py.

ⓧ I'd like to double check the full code

Here is the full worker.py for comparison:

import asyncio
import json

import redis.asyncio as aioredis


async def process_analytics():
    client = aioredis.from_url("redis://localhost:6379", decode_responses=True)
    print("Analytics worker started. Waiting for events...")

    try:
        while True:
            result = await client.brpop("analytics:queue", timeout=0)

            if result:
                _, event_data = result
                event = json.loads(event_data)
                short_code = event["short_code"]

                await client.hincrby(f"url:{short_code}", "clicks", 1)
                clicks = await client.hget(f"url:{short_code}", "clicks")
                print(f"Click processed: /{short_code} | Total clicks: {clicks}")
    except asyncio.CancelledError:
        pass
    finally:
        await client.close()


if __name__ == "__main__":
    asyncio.run(process_analytics())

For main.py, scroll up and confirm your redirect_url function now includes the json.dumps and lpush lines between the 404 check and the return statement. Everything else in main.py stays unchanged from Step 4.

Test the full producer/consumer flow

Time to prove the whole pipeline works. You will start the worker, shorten a fresh URL, click it multiple times, and watch the worker process each event in real time.

  • First, flush the test event from earlier so you start with a clean queue:
docker exec redis redis-cli DEL analytics:queue
  • Open a second terminal tab (keep the fastapi dev server running in the first).
  • Activate your virtual environment and start the worker by running:
cd ~/Desktop/url-shortener && source venv/bin/activate && python worker.py

You should see Analytics worker started. Waiting for events... printed. The worker is now blocking on BRPOP, waiting for events to arrive.

  • Open a third terminal tab. Shorten a fresh URL by running:
curl -X POST http://127.0.0.1:8000/shorten -H "Content-Type: application/json" -d '{"url": "https://example.com"}'
  • Copy the short_code from the response.
  • Trigger 3 redirects by running this command 3 times (replacing the code with yours):
curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:8000/[[SHORT_CODE="your short code"]]

Each time you run that command, switch to the worker terminal. You should see a new line printed for each click:

Click processed: /abc123 | Total clicks: 1
Click processed: /abc123 | Total clicks: 2
Click processed: /abc123 | Total clicks: 3

Before you verify the count directly in Redis, predict: what value do you expect the clicks field to hold?

  • Confirm the click count stored in the Redis hash by running (replace the code with your actual short code):
docker exec redis redis-cli HGET url:[[SHORT_CODE="your short code"]] clicks

You should see 3. The worker consumed all three events and incremented the counter each time.

Worker not printing events?

  • Make sure the worker terminal shows "Waiting for events..." before you trigger redirects. If it crashed on startup, check the import lines match exactly.
  • Confirm the FastAPI dev server reloaded after you saved the lpush change. If you saved main.py after starting the worker but before triggering redirects, events should flow through.
  • If you see a connection error, confirm Redis is still running with docker ps. Restart it if needed with docker start redis.

My analytics worker isn't processing click events from the queue. Help me debug.

Your URL shortener now has a fully decoupled analytics pipeline. The redirect stays fast because it only does an O(1) LPUSH, while the heavy work of updating counters happens in a separate process. Next up, the Secret Mission challenges you to add Redis caching and a stats endpoint on top of this foundation.

Secret mission

Add Caching and Stats

Your redirect route hits the Redis hash on every single request. Popular short URLs could get thousands of clicks per minute, and each one runs a full HGETALL. Can you add a caching layer that serves popular redirects from a fast string lookup, and expose an analytics endpoint so users can see how their links are performing?

Clean Up Your Resources

Clean Up Your Resources

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

Resources you used:

  • Redis Docker container (redis).
  • Python virtual environment and project files in the url-shortener folder on your Desktop.

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with your URL shortener.

  • Your Redis container continues running in the background and uses a small amount of memory.
  • Your url-shortener project folder stays on your Desktop for easy access.
  • To start working again later, activate your virtual environment and run fastapi dev main.py.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files and container stay intact so you can resume later.

  • Stop the FastAPI dev server by pressing Ctrl+C in its terminal.
  • Stop the analytics worker by pressing Ctrl+C in its terminal.
  • Stop the Redis container by running:
docker stop redis
  • To resume later, start Redis again with docker start redis, activate your virtual environment, and run fastapi dev main.py.

Delete - I don't want to use this again

Remove all project resources and start fresh. This permanently deletes your code and Redis data.

  • Stop the FastAPI dev server and the analytics worker by pressing Ctrl+C in each terminal.
  • Stop and remove the Redis Docker container by running:
docker stop redis && docker rm redis
  • Delete the project folder (macOS/Linux) by running:
rm -rf ~/Desktop/url-shortener
  • Delete the project folder (Windows) by running:
rmdir /s /q %USERPROFILE%\Desktop\url-shortener
  • Confirm the folder is gone by running ls ~/Desktop (macOS/Linux) or dir %USERPROFILE%\Desktop (Windows). You should no longer see url-shortener in the listing.

Nice Work!

Nice Work!

You just built a production-hardened URL shortener API where Redis powers every layer of the stack. From data storage to rate limiting to background analytics, you proved that a single tool can handle multiple infrastructure concerns when you understand its data structures.

You've learned how to:

  • Build a URL shortener API powered entirely by Redis hashes, using HSET and HGETALL to store and retrieve URL data with atomic field updates.
  • Implement a custom sliding window rate limiter using Redis sorted sets and pipelines, enforcing per-IP request limits and returning proper 429 responses with rate limit headers.
  • Decouple analytics processing using Redis lists as a producer/consumer queue, with a separate background worker consuming click events via BRPOP and incrementing counters in real time.
  • Secret Mission: Added a cache-aside pattern with a 300-second TTL to speed up popular redirects, and exposed click analytics through a /stats/{short_code} endpoint.

Ready to quiz yourself?