Module 12 · Production Engineering · Drills
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.
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.
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.
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).
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.
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"}
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.
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.
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.
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.
/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.
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.
Click a card to flip it. Say the answer out loud before you flip — that's the rep that builds storage strength.
await r.set(key, val, ex=60)INCR for a counter, not get-then-set?Tick each only if you can do it without looking: