Module 12 · Production Engineering · Drills

Drills: Caching & Background Jobs

Reading about caches and queues isn't the same as wiring one up. Type every drill yourself — run a local Redis, hit your endpoints — before you reveal the solution. The muscle memory is the point.

How to use this page Each drill is a small task. Attempt it first, run it against a real Redis (docker run -p 6379:6379 redis is enough), then click "Show solution" to compare. If yours works differently but correctly — great, that's fluency. Tick each box as you go; your progress is saved in this browser.

A · Warm-up reps Basic

Drill 1 cache-aside

Write an async get_user(uid) that follows cache-aside: check Redis, return on a hit, otherwise call fetch_from_db(uid), store the result as JSON with a 5-minute TTL, and return it.

Show solution
import json
import redis.asyncio as redis
r = redis.from_url("redis://localhost:6379", decode_responses=True)

async def get_user(uid: str):
    key = f"user:{uid}"
    cached = await r.get(key)
    if cached is not None:
        return json.loads(cached)        # HIT
    user = await fetch_from_db(uid)        # MISS — do the work
    await r.set(key, json.dumps(user), ex=300)
    return user

Drill 2 cache key

Write cache_key(doc_id, question) that builds a stable, fixed-length key by hashing the normalised inputs with SHA-256 (take the first 16 hex chars).

Show solution
import hashlib

def cache_key(doc_id: str, question: str) -> str:
    raw = f"{doc_id}:{question.strip().lower()}"
    digest = hashlib.sha256(raw.encode()).hexdigest()[:16]
    return f"answer:{digest}"

Normalising (strip().lower()) means "Refund policy?" and " refund policy? " hit the same key — more cache hits, less spend.

Drill 3 redis types

Using Redis directly: atomically increment a counter asks:total, then store a job record as a hash under job:abc with status="queued" and a 1-day TTL.

Show solution
await r.incr("asks:total")                 # atomic counter

await r.hset("job:abc", mapping={"status": "queued"})
await r.expire("job:abc", 86400)         # clean up after a day
print(await r.hgetall("job:abc"))         # {"status": "queued"}

B · Stretch Intermediate

Drill 4 enqueue

Write a FastAPI /ingest endpoint that saves the upload, enqueues an ARQ task named "ingest_pdf" with the doc_id, and returns the job id with a 202 status.

Show solution
from arq import create_pool
from arq.connections import RedisSettings

@app.post("/ingest", status_code=202)
async def ingest(file: UploadFile):
    doc_id = await save_raw(file)
    pool = await create_pool(RedisSettings())
    job = await pool.enqueue_job("ingest_pdf", doc_id)
    return {"job_id": job.job_id, "doc_id": doc_id}

The endpoint does the minimum and returns in milliseconds. 202 Accepted is the honest status code: "I took your request, it's not done yet".

Drill 5 worker task

Write the ARQ worker task ingest_pdf(ctx, doc_id) that parses, chunks, embeds each chunk, stores it, and returns the chunk count. Then write the WorkerSettings that registers it with a 10-minute timeout and 3 retries.

Show solution
from arq.connections import RedisSettings

async def ingest_pdf(ctx, doc_id: str):
    text = await parse_pdf(doc_id)
    chunks = chunk(text)
    for c in chunks:
        vec = await embed_cached(c)
        await store(doc_id, c, vec)
    return {"chunks": len(chunks)}

class WorkerSettings:
    functions = [ingest_pdf]
    redis_settings = RedisSettings(host="localhost", port=6379)
    job_timeout = 600
    max_tries = 3

# run it:  arq worker.WorkerSettings

Drill 6 idempotency

Add a retry-safe guard to ingest_pdf: if the doc is already marked ingested in Redis, skip; otherwise clear any existing chunks before re-inserting, and set the "done" flag at the end.

Show solution
async def ingest_pdf(ctx, doc_id: str):
    if await r.get(f"ingested:{doc_id}"):
        return {"skipped": True}        # already done — retry is a no-op
    await clear_chunks(doc_id)              # safe if a prior run half-finished
    chunks = chunk(await parse_pdf(doc_id))
    for c in chunks:
        await store(doc_id, c, await embed_cached(c))
    await r.set(f"ingested:{doc_id}", "1", ex=86400)
    return {"chunks": len(chunks)}

Ask of every task: "what if this runs twice?" The flag plus the clear make a retry harmless instead of a duplicate-data bug.

C · Build challenge Build

Mini-project Build DocChat's full async ingestion job with status polling. The /ingest endpoint enqueues the job and returns an id. The worker task is idempotent, embeds with a cache, and writes progress into a Redis hash as it goes. A /jobs/{job_id} endpoint reports status and progress so the frontend can poll it to a finished bar. This is the exact production shape interviewers ask you to draw.

Build · async ingestion + polling

Wire the endpoint, the worker task, and the status route together.

Show solution
main.py
from fastapi import FastAPI, UploadFile
from arq import create_pool
from arq.connections import RedisSettings
from arq.jobs import Job
import redis.asyncio as redis

app = FastAPI()
r = redis.from_url("redis://localhost:6379", decode_responses=True)

@app.post("/ingest", status_code=202)
async def ingest(file: UploadFile):
    doc_id = await save_raw(file)
    pool = await create_pool(RedisSettings())
    job = await pool.enqueue_job("ingest_pdf", doc_id)
    return {"job_id": job.job_id}

@app.get("/jobs/{job_id}")
async def job_status(job_id: str):
    pool = await create_pool(RedisSettings())
    st = await Job(job_id, pool).status()
    progress = await r.hget(f"job:{job_id}", "progress")
    return {"status": str(st), "progress": progress or "0"}
worker.py
from arq.connections import RedisSettings
import redis.asyncio as redis
r = redis.from_url("redis://localhost:6379", decode_responses=True)

async def ingest_pdf(ctx, doc_id: str):
    job_id = ctx["job_id"]
    if await r.get(f"ingested:{doc_id}"):     # idempotency guard
        return {"skipped": True}
    await clear_chunks(doc_id)
    chunks = chunk(await parse_pdf(doc_id))
    total = len(chunks)
    for i, c in enumerate(chunks, start=1):
        await store(doc_id, c, await embed_cached(c))
        await r.hset(f"job:{job_id}", "progress", str(i / total))
    await r.set(f"ingested:{doc_id}", "1", ex=86400)
    return {"chunks": total}

class WorkerSettings:
    functions = [ingest_pdf]
    redis_settings = RedisSettings(host="localhost", port=6379)
    job_timeout = 600
    max_tries = 3

The frontend now polls GET /jobs/{id} every second, reads progress from 0.0 → 1.0, and shows a real bar. The upload returned instantly — no timeout, no spinner of doom.

D · Rapid recall Flashcards

Click a card to flip it. Say the answer out loud before you flip — that's the rep that builds storage strength.

The four steps of cache-aside?
Get → on miss compute → set with TTL → return.
click to flip
Set a key with a 60-second TTL in redis-py?
await r.set(key, val, ex=60)
click to flip
Why use INCR for a counter, not get-then-set?
It's atomic — no lost-update race between requests.
click to flip
Why is BackgroundTasks wrong for ingestion?
In-process, no retry, work is lost on a crash.
click to flip
What makes a task idempotent?
Running it twice has the same effect as running it once.
click to flip
SSE vs WebSocket for a progress bar?
SSE — one-way server→client is all you need.
click to flip

E · Self-check before moving on

Tick each only if you can do it without looking:

Next All ticked? Your slow work is queued and your repeat work is cached — but how will you know when a job fails at 3am or a cache hit rate quietly collapses? Next we make the system visible: Lesson 12.3 — Observability: Logging, Metrics & Tracing.