Module 12 · Production Engineering · Deep Dive

Caching & Background Jobs

Make the fast things instant and move the slow things off the request — the two moves that separate a demo from a service real users can lean on.

BasicIntermediateBuild

1 · Why cache at all

A cache is a small, fast store that holds the answer to a question you've already answered, so you don't have to compute it again. You reach for one when work is repeated and expensive. Three forces push you there:

• Latency — reading a value from memory takes microseconds; recomputing it (a database join, an LLM call) takes hundreds of milliseconds or seconds. Users feel the difference instantly.
• Cost — every embedding call and every answer from the model is money. In DocChat, two users asking "what is the refund policy?" should cost you one LLM call, not two.
• Load — a cache absorbs traffic that would otherwise hammer your database or a rate-limited upstream API. It's the cheapest way to survive a spike.

The trade is always the same: you exchange freshness for speed. A cached value can be stale. The whole craft of caching is deciding how stale is acceptable and how you'll fix it when it isn't.

2 · The cache-aside pattern

By far the most common pattern, and the one you should reach for by default. The application — not the cache — is in charge. The logic is always these four steps:

# cache-aside (a.k.a. lazy loading)
async def get_value(key):
    cached = await cache.get(key)        # 1. look in the cache
    if cached is not None:
        return cached                    # 2. HIT — return it, we're done
    value = await compute_expensive(key) # 3. MISS — do the real work
    await cache.set(key, value, ttl=3600) # 4. store it for next time
    return value

It's "aside" because the cache sits to the side of your data store, not in front of it. The first request for a key always misses and pays full price; every request after that (until the TTL expires) is a fast hit. This lazy behaviour is a feature — you only ever cache things someone actually asked for.

TTL the easy half of invalidation

A TTL (time-to-live) tells the cache to forget a value after N seconds. It's your cheapest invalidation strategy: set a value to live for an hour, accept that it might be up to an hour stale, and move on. For derived data like an LLM answer, that's usually fine.

Invalidation the genuinely hard half

There are, as the saying goes, only two hard things in computer science — and cache invalidation is one of them. The problem: when the underlying truth changes, every cached copy is now a lie. If a user re-uploads a corrected version of their document, the cached answers from the old version are wrong, and the TTL might not expire for another 50 minutes.

Two honest strategies:

• Expire — lean on TTLs and tolerate bounded staleness. Simple, robust, good enough for most derived data.
• Invalidate on write — when you change the source of truth, actively delete (or overwrite) the affected cache keys in the same code path. Correct, but you must remember every key the change touches, which is exactly where bugs live.

3 · Redis with async redis-py

Redis is an in-memory data store you'll use as your cache, your shared session/rate-limit store, and the backbone of your job queue. It's a separate process (or managed service) your app talks to over the network. In 2026 the standard Python client is redis-py, which ships first-class async support — perfect for FastAPI.

cache.py

import redis.asyncio as redis

# one shared pool for the whole app; decode bytes to str for us
r = redis.from_url("redis://localhost:6379", decode_responses=True)

async def demo():
    await r.set("greeting", "hello", ex=60)  # ex = TTL in seconds
    val = await r.get("greeting")            # "hello"
    ttl = await r.ttl("greeting")            # ~60, counts down

Everything in Redis is keyed by a string, but the value can be one of several data types. You'll use four constantly:

# strings — a counter (atomic, no read-modify-write race)
await r.incr("asks:total")

# hashes — a small record under one key
await r.hset("job:abc", mapping={"status": "running", "progress": "0"})
await r.hgetall("job:abc")   # {"status": "running", "progress": "0"}

# sets — uniqueness & fast membership
await r.sadd("user:7:docs", "doc-12")
await r.sismember("user:7:docs", "doc-12")   # True

# sorted sets — ranked by a numeric score
await r.zadd("leaderboard", {"sam": 42})
await r.zrange("leaderboard", 0, -1, desc=True)

# EXPIRE sets a TTL on any existing key
await r.expire("job:abc", 86400)   # clean up after a day

4 · Caching DocChat's expensive work

DocChat has two costs worth caching, and both are deterministic — same input, same output — which is exactly what makes them safe to cache.

1 Repeated question → answer pairs

If two users ask the same question of the same document, the answer is identical. Cache it. The key must capture everything the answer depends on: the document (with a version), and the normalised question.

ask_cache.py

import hashlib, json

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

async def ask(doc_id, doc_version, question):
    key = answer_key(doc_id, doc_version, question)
    cached = await r.get(key)
    if cached:
        return json.loads(cached)            # cache HIT — instant, free
    answer = await run_rag_pipeline(doc_id, question)  # slow + costs money
    await r.set(key, json.dumps(answer), ex=3600)
    return answer

Why hash the input into the key? Because raw questions can be long, contain characters Redis would rather not see, and leak content into your key space. A sha256 digest gives you a fixed-length, opaque, collision-safe key from any input. Note the doc_version in the key — re-upload the document, bump the version, and every old answer key is naturally orphaned (and expires on its own).

2 Embeddings

Embedding a chunk of text is a model call. The same chunk always embeds to the same vector, so you never need to compute it twice — even across documents that share boilerplate. Cache by a hash of the chunk text:

async def embed_cached(text: str) -> list[float]:
    key = "emb:" + hashlib.sha256(text.encode()).hexdigest()[:16]
    hit = await r.get(key)
    if hit:
        return json.loads(hit)
    vec = await embedding_model.embed(text)
    await r.set(key, json.dumps(vec), ex=86400 * 7)   # a week
    return vec

On a re-ingest of a lightly-edited document, most chunks are unchanged, so most embeds become free cache hits. That's a real bill reduction, not a micro-optimisation.

5 · Redis as shared state across instances

The moment you run more than one copy of your app — and in production you always do — anything stored inside a single process becomes a lie. An in-memory Python dict used as a cache, a rate-limit counter, or a session store is invisible to the other instances. Requests bounce between instances behind a load balancer, so a counter on instance A says "3" while instance B says "1".

Redis is the shared brain that fixes this. Two classic uses:

# RATE LIMITING — one shared counter, atomic increment
async def allow(user_id: str, limit=100) -> bool:
    key = f"rl:{user_id}:{int(time.time()) // 60}"   # per-minute window
    count = await r.incr(key)
    if count == 1:
        await r.expire(key, 60)        # first hit sets the window's TTL
    return count <= limit

# SESSIONS — readable from any instance
await r.set(f"session:{token}", json.dumps(user), ex=3600)
session = await r.get(f"session:{token}")

6 · Why you need background jobs

Caching handles repeated work. Background jobs handle slow work. The rule is simple and non-negotiable: an HTTP request should return in well under a second. Anything longer ties up a worker, frustrates the user, and eventually trips a timeout somewhere — the browser, your reverse proxy, or the load balancer (often a hard 30 or 60 seconds).

DocChat's ingestion is the textbook case. Uploading a 200-page PDF means: parse it, split it into hundreds of chunks, embed every chunk (hundreds of model calls), and write them all to the vector store. That's tens of seconds to minutes. You cannot do it inside the upload request. If you try, the user waits forever, a timeout kills the connection mid-write, and you're left with half-ingested data.

The pattern that fixes it: the endpoint does the minimum (validate, save the raw file, create a job), returns a job id immediately, and the heavy lifting happens in a separate worker process. The frontend then polls or subscribes for progress.

FastAPI BackgroundTasks the trivial option

FastAPI ships BackgroundTasks: schedule a function to run after the response is sent. It's genuinely useful — for trivial, fire-and-forget work.

from fastapi import BackgroundTasks

@app.post("/contact")
async def contact(msg: Message, bg: BackgroundTasks):
    bg.add_task(send_email, msg)     # runs after the response
    return {"ok": True}

7 · A real task queue with ARQ

A task queue solves everything BackgroundTasks can't. You push a job description onto a queue (backed by Redis); separate worker processes — on the same box or different machines — pull jobs off and run them. Jobs survive restarts, can be retried, report status, and scale horizontally just by running more workers.

In 2026 the natural fit for an async FastAPI app is ARQ — a lightweight async task queue built directly on redis-py. Your tasks are async def functions, the same flavour as your routes. (The heavyweight, battle-tested alternative is Celery — more on that below.)

1 Define the task and the worker

worker.py

from arq import create_pool
from arq.connections import RedisSettings

async def ingest_pdf(ctx, doc_id: str):
    """A task is just an async function whose first arg is ctx."""
    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]              # tasks this worker can run
    redis_settings = RedisSettings(host="localhost", port=6379)
    job_timeout = 600                     # kill a runaway job after 10 min
    max_tries = 3                         # retry on failure

You run the worker as its own process — separate from uvicorn:

arq worker.WorkerSettings

2 Enqueue from the endpoint & get a job id

main.py

@app.post("/ingest", status_code=202)
async def ingest(file: UploadFile):
    doc_id = await save_raw(file)            # fast: just store the bytes
    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 returns in milliseconds with a 202 Accepted and a job id. The actual ingestion runs later, in the worker.

3 Poll job status

from arq.jobs import Job

@app.get("/jobs/{job_id}")
async def job_status(job_id: str):
    pool = await create_pool(RedisSettings())
    job = Job(job_id, pool)
    status = await job.status()           # queued / in_progress / complete
    result = await job.result(timeout=0) if status == "complete" else None
    return {"status": str(status), "result": result}

8 · Retries, idempotency & progress

A queue that runs a task once on a sunny day is easy. Production is rainy: workers crash, networks blip, jobs get retried. Three things make jobs trustworthy.

Retries & timeouts

ARQ retries a failed task up to max_tries and enforces job_timeout so a stuck job can't run forever. Retries are a gift — but they create a new problem: a task might run more than once.

Idempotency the property that makes retries safe

A task is idempotent if running it twice has the same effect as running it once. Without this, a retry of ingest_pdf would embed and insert every chunk a second time — duplicate vectors, doubled bill, corrupted results. Guard against it. The cleanest approach is a Redis flag that marks a document as done:

async def ingest_pdf(ctx, doc_id: str):
    # idempotency guard: skip if already ingested
    if await r.get(f"ingested:{doc_id}"):
        return {"skipped": True}
    await clear_chunks(doc_id)          # also safe: wipe before re-insert
    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)}

Scheduled / cron jobs

Some work isn't triggered by a user at all — nightly cleanup of expired sessions, a daily re-index, pruning orphaned vectors. ARQ runs these on a schedule with cron jobs declared in the worker:

from arq import cron

class WorkerSettings:
    functions = [ingest_pdf]
    cron_jobs = [
        cron(cleanup_expired, hour=3, minute=0),   # every day at 03:00
    ]

Reporting progress to the frontend

Polling /jobs/{id} every second works and is the simplest thing that does. For a smoother bar, push progress to the browser. The task writes progress into Redis; the API streams it out. Two transports:

• Server-Sent Events (SSE) — one-way server→client stream over plain HTTP. Ideal for "ingestion is 40% done". Simple, auto-reconnects, no extra protocol.
• WebSockets — full duplex. Reach for it only when the client also needs to push (live chat, collaborative editing). Overkill for a progress bar.

# task writes progress as it goes
await r.hset(f"job:{job_id}", "progress", str(done / total))

# SSE endpoint streams it to the browser
from sse_starlette.sse import EventSourceResponse

@app.get("/ingest/{job_id}/stream")
async def stream(job_id: str):
    async def gen():
        while True:
            p = await r.hget(f"job:{job_id}", "progress")
            yield {"data": p or "0"}
            if p == "1.0":
                break
            await asyncio.sleep(1)
    return EventSourceResponse(gen())

Celery the heavyweight alternative

For completeness: Celery is the long-established Python task queue. It's more mature and feature-rich than ARQ — complex routing, multiple broker backends (Redis or RabbitMQ), a huge ecosystem — but it's heavier, historically sync-first, and more configuration to stand up. The decision rule: a modern async FastAPI app with straightforward background work? ARQ. A large system that needs elaborate routing, mixed languages, or you already run RabbitMQ? Celery. Both put Redis at the centre; the concepts you learned here transfer directly.

9 · Build it

The endpoint and the answer cache, wired together:

main.py

import json, hashlib
import redis.asyncio as redis
from fastapi import FastAPI, UploadFile
from arq import create_pool
from arq.connections import RedisSettings
from arq.jobs import Job

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)                  # fast — bytes to disk
    pool = await create_pool(RedisSettings())
    job = await pool.enqueue_job("ingest_pdf", doc_id)
    return {"job_id": job.job_id, "doc_id": doc_id}

@app.get("/jobs/{job_id}")
async def status(job_id: str):
    pool = await create_pool(RedisSettings())
    job = Job(job_id, pool)
    st = await job.status()
    return {"status": str(st)}

@app.post("/ask")
async def ask(doc_id: str, doc_version: int, question: str):
    raw = f"{doc_id}:{doc_version}:{question.strip().lower()}"
    key = "answer:" + hashlib.sha256(raw.encode()).hexdigest()[:16]
    cached = await r.get(key)
    if cached:
        return {"answer": json.loads(cached), "cached": True}
    answer = await run_rag_pipeline(doc_id, question)
    await r.set(key, json.dumps(answer), ex=3600)
    return {"answer": answer, "cached": False}

Build the worker side (with the idempotency guard and progress writes) in the drills. The shape you've just written — accept fast, queue the slow work, return an id, cache the repeat — is the single most reusable production pattern in the course.

10 · Check yourself

Answer from memory — retrieval is what moves this from "I read it" to "I know it".