Master Caching Patterns in FastAPI
Build a two-layer caching system in FastAPI using memory and Redis.
Introduction
30 Second Summary
Every time a website loads instantly, caching is working behind the scenes. Without it, every single request hits the database directly, making users wait hundreds of milliseconds for data that hasn't even changed.
In this project, you will build a Product Catalog API with FastAPI that progressively adds caching layers. You'll start with a slow API that simulates database latency, then add in-memory caching and Redis caching one layer at a time, benchmarking the improvement at each stage.
What You'll Build
A FastAPI application with a two-layer caching system (in-memory + Redis) that reduces API response times from 200ms+ to under 2ms, with built-in latency timing and cache-hit metadata in every response.
By the end of this project, you'll have:
- A working FastAPI API with in-memory caching using cachetools TTLCache that serves repeated requests in under 1ms.
- A Redis cache layer via Upstash that enables distributed caching shared across multiple processes and server restarts.
- Production-ready cache invalidation patterns (write-through and manual clearing) that keep cached data consistent with source data.
- Secret Mission: Add cache stampede protection using Redis distributed locking to handle thundering herd scenarios.
Are there any prerequisites?
You should be comfortable writing FastAPI endpoints and have a basic understanding of async/await in Python. No prior caching experience is needed.
Set Up the Project
Before you can build caching layers, you need a working FastAPI application that starts up cleanly and connects to external services on launch.
In this step, you will create the application skeleton with a lifespan function that manages startup and shutdown logic for your Redis connection.
In this step, get ready to:
- Create main.py with a minimal FastAPI app skeleton using the lifespan pattern.
- Verify the app runs with fastapi dev main.py.
Create the app skeleton
The lifespan pattern lets you run code when your app starts and when it shuts down. You will use it to open and close the Redis connection gracefully.
- In VS Code, create a new file called main.py in your project folder.
- Add the imports and configuration variables by pasting the following code into main.py:
import asyncio
import json
import os
import time
from cachetools import TTLCache
from fastapi import FastAPI, HTTPException
from redis.asyncio import Redis
from contextlib import asynccontextmanager
REDIS_URL = os.getenv("REDIS_URL", "")
redis_client: Redis | None = NoneWhat does this code do?
The imports pull in everything the caching system needs: asyncio for async sleep, json for serializing data to Redis, time for latency measurement, and TTLCache for in-memory caching.
REDIS_URL reads the connection string from an environment variable. If it is empty, the app runs without Redis. redis_client is a module-level placeholder that the lifespan function will fill in at startup.
- Add the lifespan function and app instantiation below the configuration variables:
@asynccontextmanager
async def lifespan(app: FastAPI):
global redis_client
if REDIS_URL:
redis_client = Redis.from_url(REDIS_URL, decode_responses=True)
try:
await redis_client.ping()
print("Connected to Redis")
except Exception as e:
print(f"Redis connection failed: {e}. Running without Redis.")
redis_client = None
else:
print("No REDIS_URL set. Running with in-memory cache only.")
yield
if redis_client:
await redis_client.close()
app = FastAPI(title="Product Catalog API - Caching Demo", lifespan=lifespan)What does this code do?
The lifespan function is an async context manager. Everything before yield runs at startup. Everything after yield runs at shutdown.
At startup, if a REDIS_URL environment variable is set, the function creates a Redis client and tests the connection with a ping(). If the connection fails, it falls back gracefully to running without Redis.
At shutdown, it closes the Redis connection cleanly so no resources leak.
- Save main.py.
Seeing a red underline on Redis | None?
The Redis | None syntax requires Python 3.10 or later. If your editor shows an error, confirm your virtual environment uses a recent Python version. Help me check my Python version and fix type hint errors.
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py before moving on.
ⓧ I'd like to double check the full code
import asyncio
import json
import os
import time
from cachetools import TTLCache
from fastapi import FastAPI, HTTPException
from redis.asyncio import Redis
from contextlib import asynccontextmanager
REDIS_URL = os.getenv("REDIS_URL", "")
redis_client: Redis | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global redis_client
if REDIS_URL:
redis_client = Redis.from_url(REDIS_URL, decode_responses=True)
try:
await redis_client.ping()
print("Connected to Redis")
except Exception as e:
print(f"Redis connection failed: {e}. Running without Redis.")
redis_client = None
else:
print("No REDIS_URL set. Running with in-memory cache only.")
yield
if redis_client:
await redis_client.close()
app = FastAPI(title="Product Catalog API - Caching Demo", lifespan=lifespan)Verify the app runs
Now that the skeleton is in place, you can start the development server to confirm everything imports correctly and the lifespan function executes.
- Start the FastAPI development server by running this command in your terminal:
fastapi dev main.py- Look for the following output in your terminal:
No REDIS_URL set. Running with in-memory cache only.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Application startup complete.What should I see?
The key line is No REDIS_URL set. Running with in-memory cache only. which confirms your lifespan function ran at startup. You have not set a Redis connection yet, so this is expected.
The server is now running at http://127.0.0.1:8000. You can open that URL in your browser to see the default FastAPI response. The auto-generated docs are available at http://127.0.0.1:8000/docs.
Command not found or module errors?
Make sure your virtual environment is activated before running the command. You should see the environment name in your terminal prompt.
If you see ModuleNotFoundError, your dependencies may not have installed into the active environment. Reactivate your virtual environment and reinstall. Help me troubleshoot my FastAPI dev server not starting.
Your app skeleton is running. Next up, you will build the product endpoints that simulate slow database queries, giving you something concrete to speed up with caching.
Build the Uncached API
Your FastAPI app is running, but it doesn't do anything yet. Before adding caching, you need a slow API to cache.
In this step, you'll build a product catalog API that simulates real-world database latency. Every request will take 200ms or more. This baseline gives you a clear "before" measurement so you can see exactly how much caching improves things later.
In this step, get ready to:
- Add a simulated product database with artificial latency.
- Create two GET endpoints that return products with timing metadata.
- Test the endpoints and observe consistent 200ms+ response times.
Add the simulated product database
Real APIs query databases, and database queries take time. Instead of setting up an actual database, you'll use a Python dictionary and asyncio.sleep() to simulate that latency.
- In main.py, find the line that says REDIS_URL = os.getenv("REDIS_URL", "").
- Add the following product data ABOVE that line:
# --- Simulated Database ---
PRODUCTS_DB = {
"1": {"id": "1", "name": "Wireless Headphones", "price": 79.99, "category": "electronics", "stock": 150},
"2": {"id": "2", "name": "Running Shoes", "price": 129.99, "category": "sports", "stock": 75},
"3": {"id": "3", "name": "Coffee Maker", "price": 49.99, "category": "kitchen", "stock": 200},
"4": {"id": "4", "name": "Yoga Mat", "price": 29.99, "category": "sports", "stock": 300},
"5": {"id": "5", "name": "Bluetooth Speaker", "price": 59.99, "category": "electronics", "stock": 90},
}What does this code do?
PRODUCTS_DB is a dictionary that acts as your fake database. Each key is a product ID, and each value contains the product details (name, price, category, stock).
In a real application, this data would live in PostgreSQL, MongoDB, or another database. The dictionary lets you focus on caching patterns without database setup.
- Below the PRODUCTS_DB dictionary you just added, add these async functions that simulate slow database queries:
async def simulate_db_query(product_id: str) -> dict | None:
"""Simulates a slow database query with 200ms latency."""
await asyncio.sleep(0.2)
return PRODUCTS_DB.get(product_id)
async def simulate_db_query_all() -> list[dict]:
"""Simulates fetching all products with 500ms latency."""
await asyncio.sleep(0.5)
return list(PRODUCTS_DB.values())
# --- Redis Client ---What does this code do?
- simulate_db_query() waits 200ms (via asyncio.sleep(0.2)) then returns a single product. This mimics querying one row from a database.
- simulate_db_query_all() waits 500ms then returns all products. Listing all rows is slower than fetching one, so the latency is higher.
- The # --- Redis Client --- comment is a section divider. It labels the existing Redis code below it for readability.
- Save main.py.
Seeing a syntax error in the terminal?
Make sure the PRODUCTS_DB dictionary and both functions are placed ABOVE the REDIS_URL line. The two simulate functions reference PRODUCTS_DB, so the dictionary must appear first.
Check that you haven't accidentally placed code inside the lifespan function. Everything in this substep should be at the top level of the file (no indentation).
Help me debug a syntax error in my FastAPI product database code
Create the GET endpoints with timing metadata
Now you need endpoints that call the slow database functions. Each response will include a latency_ms field so you can see exactly how long every request takes.
- In main.py, scroll to the bottom of the file. Find the line that says app = FastAPI(title="Product Catalog API - Caching Demo", lifespan=lifespan).
- Add the single-product endpoint below that line:
@app.get("/products/{product_id}")
async def get_product(product_id: str):
start = time.perf_counter()
product = await simulate_db_query(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
elapsed = time.perf_counter() - start
return {
"data": product,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}What does this code do?
- time.perf_counter() records a high-precision timestamp before and after the database call. The difference gives you the exact latency.
- The response always says "cache": "miss" because there's no cache yet. Later, this field will change to "memory" or "redis" when caching kicks in.
- latency_ms converts the elapsed seconds to milliseconds. This gives you a clear number to compare across cache states.
- Below the get_product endpoint, add the list-all-products endpoint:
@app.get("/products")
async def get_all_products():
start = time.perf_counter()
products = await simulate_db_query_all()
elapsed = time.perf_counter() - start
return {
"data": products,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}What does this code do?
This endpoint follows the same pattern: start a timer, call the database function, measure elapsed time, and return the data with timing metadata. The only difference is it calls simulate_db_query_all() which has 500ms latency instead of 200ms.
- Save main.py. The dev server auto-reloads your changes.
Server showing an import error?
Check that you haven't accidentally deleted any of the import lines at the top of main.py. The endpoints use time, HTTPException, and asyncio, which should already be imported from Step 1.
Help me fix an import error in my FastAPI app
✔️ Awesome, I've got everything!
Great. Double check you've saved your file, then move on to testing.
ⓧ I'd like to double check the full code
import asyncio
import json
import os
import time
from cachetools import TTLCache
from fastapi import FastAPI, HTTPException
from redis.asyncio import Redis
from contextlib import asynccontextmanager
# --- Simulated Database ---
PRODUCTS_DB = {
"1": {"id": "1", "name": "Wireless Headphones", "price": 79.99, "category": "electronics", "stock": 150},
"2": {"id": "2", "name": "Running Shoes", "price": 129.99, "category": "sports", "stock": 75},
"3": {"id": "3", "name": "Coffee Maker", "price": 49.99, "category": "kitchen", "stock": 200},
"4": {"id": "4", "name": "Yoga Mat", "price": 29.99, "category": "sports", "stock": 300},
"5": {"id": "5", "name": "Bluetooth Speaker", "price": 59.99, "category": "electronics", "stock": 90},
}
async def simulate_db_query(product_id: str) -> dict | None:
"""Simulates a slow database query with 200ms latency."""
await asyncio.sleep(0.2)
return PRODUCTS_DB.get(product_id)
async def simulate_db_query_all() -> list[dict]:
"""Simulates fetching all products with 500ms latency."""
await asyncio.sleep(0.5)
return list(PRODUCTS_DB.values())
# --- Redis Client ---
REDIS_URL = os.getenv("REDIS_URL", "")
redis_client: Redis | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global redis_client
if REDIS_URL:
redis_client = Redis.from_url(REDIS_URL, decode_responses=True)
try:
await redis_client.ping()
print("Connected to Redis")
except Exception as e:
print(f"Redis connection failed: {e}. Running without Redis.")
redis_client = None
else:
print("No REDIS_URL set. Running with in-memory cache only.")
yield
if redis_client:
await redis_client.close()
app = FastAPI(title="Product Catalog API - Caching Demo", lifespan=lifespan)
@app.get("/products/{product_id}")
async def get_product(product_id: str):
start = time.perf_counter()
product = await simulate_db_query(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
elapsed = time.perf_counter() - start
return {
"data": product,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}
@app.get("/products")
async def get_all_products():
start = time.perf_counter()
products = await simulate_db_query_all()
elapsed = time.perf_counter() - start
return {
"data": products,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}Test the endpoints and observe the latency
Your dev server should still be running from Step 1. If you stopped it, restart it with fastapi dev main.py.
- Open your browser and navigate to http://127.0.0.1:8000/products/1.
You should see a JSON response like this:
{
"data": {
"id": "1",
"name": "Wireless Headphones",
"price": 79.99,
"category": "electronics",
"stock": 150
},
"cache": "miss",
"latency_ms": 200.45
}What should I see?
The latency_ms value should be approximately 200 or slightly above. The exact number varies by machine, but it will always be 200ms+ because of the asyncio.sleep(0.2) inside simulate_db_query().
The "cache": "miss" field tells you no cache was used. Refresh the page multiple times. Notice that the latency stays at 200ms+ every single time. There's no improvement on repeated requests.
- Now navigate to http://127.0.0.1:8000/products to test the list endpoint.
This response returns all five products. The latency_ms value should be approximately 500ms+ because simulate_db_query_all() uses a longer delay.
- Try requesting a product that doesn't exist by navigating to http://127.0.0.1:8000/products/99.
You should see a 404 response with the message "Product not found". This confirms your error handling works.
Getting a "Not Found" error on /products/1?
If you see a generic 404 (not your custom "Product not found" message), the endpoint might not be registered. Check your terminal for error messages. The dev server prints errors when it fails to load your code.
Make sure your endpoint decorators (@app.get("/products/{product_id}")) are at the top level of the file, not indented inside another function.
Help me debug why my /products endpoint isn't working
Why build a slow API on purpose?
This 200ms latency is your baseline. Every single request hits the "database" and waits. In the next steps, you'll add caching layers that drop this to under 2ms for repeated requests.
By building the uncached version first, you'll have hard numbers to compare against. You'll know exactly how much each caching layer improves performance.
Your API works, but it's painfully slow on every request. Even repeated requests for the same product take 200ms+. Next up, you'll add an in-memory cache that makes repeated requests nearly instant.
Add In-Memory Caching
In the previous step, you confirmed that every request to your API takes 200ms or more. That latency adds up fast when real users are refreshing the same product pages repeatedly.
Most product data does not change between requests. If you store a copy of recent results in your application's memory, repeated requests can skip the database entirely. In this step, you will add an in-memory cache using cachetools and implement the cache-aside pattern to drop response times from 200ms to under 1ms.
In this step, get ready to:
- Add a TTLCache that stores query results for 30 seconds.
- Implement the cache-aside pattern in both product endpoints.
- Test that repeated requests drop from 200ms to under 1ms.
Add a TTLCache and implement cache-aside for the single product endpoint
The cache-aside pattern works in three stages: check the cache first, on a miss query the database, then store the result in the cache before returning it. The next request for the same data skips the database entirely.
- In main.py, find the # --- Redis Client --- comment.
- Add the following code directly above that comment to create your in-memory cache:
# --- In-Memory Cache Layer ---
memory_cache = TTLCache(maxsize=100, ttl=30)What is TTLCache?
TTLCache is a dictionary-like object from the cachetools library. It automatically evicts entries after a time-to-live (TTL) period expires.
Here, maxsize=100 means the cache holds at most 100 entries before evicting the oldest. ttl=30 means each entry expires after 30 seconds, forcing a fresh database query.
Now you will update the single product endpoint to check the cache before querying the database.
- Find the get_product function (the @app.get("/products/{product_id}") endpoint).
- Replace the entire function body with this cache-aside implementation:
@app.get("/products/{product_id}")
async def get_product(product_id: str):
start = time.perf_counter()
cache_key = f"product:{product_id}"
# Layer 1: Check in-memory cache
if cache_key in memory_cache:
elapsed = time.perf_counter() - start
return {
"data": memory_cache[cache_key],
"cache": "memory",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 2: Database (cache miss)
product = await simulate_db_query(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
memory_cache[cache_key] = product
elapsed = time.perf_counter() - start
return {
"data": product,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}What does this code do?
- cache_key = f"product:{product_id}" builds a unique key per product so each product is cached independently.
- The if cache_key in memory_cache check is the "aside" part. If the data is already cached, return it immediately without touching the database.
- On a miss, the function queries the database, stores the result with memory_cache[cache_key] = product, and then returns it.
- The response includes "cache": "memory" on a hit or "cache": "miss" so you can observe exactly what happened.
- Save main.py.
Seeing a NameError for memory_cache?
Make sure the memory_cache = TTLCache(maxsize=100, ttl=30) line appears above the endpoint functions, between the simulated database section and the Redis client section.
help me fix a NameError for memory_cache in my FastAPI app
Cache the list endpoint
The /products endpoint takes 500ms because it simulates fetching all products at once. The same cache-aside pattern applies here.
- Find the get_all_products function (the @app.get("/products") endpoint).
- Replace the entire function body with this cache-aside version:
@app.get("/products")
async def get_all_products():
start = time.perf_counter()
cache_key = "products:all"
# Layer 1: Check in-memory cache
if cache_key in memory_cache:
elapsed = time.perf_counter() - start
return {
"data": memory_cache[cache_key],
"cache": "memory",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 2: Database (cache miss)
products = await simulate_db_query_all()
memory_cache[cache_key] = products
elapsed = time.perf_counter() - start
return {
"data": products,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}What does this code do?
This follows the same cache-aside pattern as the single product endpoint. The key "products:all" caches the entire product list as one entry.
On a hit, the endpoint returns the full list from memory in under 1ms instead of waiting 500ms for the simulated database.
- Save main.py.
Why is in-memory caching limited?
This cache lives inside your Python process. If you restart the server, the cache is gone. If you run multiple workers (common in production), each worker has its own separate cache that cannot share data.
For a single-server learning project, this is fine. In the next step, you will add Redis as a shared cache layer that solves both problems.
Server not reloading after save?
The fastapi dev command auto-reloads when it detects file changes. If it stops responding, press Ctrl+C in your terminal and re-run fastapi dev main.py.
help me fix auto-reload issues with fastapi dev
✔️ Awesome, I've got everything!
Great. Make sure you have saved main.py with both updated endpoints.
ⓧ I'd like to double check the full code
Here is the full main.py as it should look at this point:
import asyncio
import json
import os
import time
from cachetools import TTLCache
from fastapi import FastAPI, HTTPException
from redis.asyncio import Redis
from contextlib import asynccontextmanager
# --- Simulated Database ---
PRODUCTS_DB = {
"1": {"id": "1", "name": "Wireless Headphones", "price": 79.99, "category": "electronics", "stock": 150},
"2": {"id": "2", "name": "Running Shoes", "price": 129.99, "category": "sports", "stock": 75},
"3": {"id": "3", "name": "Coffee Maker", "price": 49.99, "category": "kitchen", "stock": 200},
"4": {"id": "4", "name": "Yoga Mat", "price": 29.99, "category": "sports", "stock": 300},
"5": {"id": "5", "name": "Bluetooth Speaker", "price": 59.99, "category": "electronics", "stock": 90},
}
async def simulate_db_query(product_id: str) -> dict | None:
"""Simulates a slow database query with 200ms latency."""
await asyncio.sleep(0.2)
return PRODUCTS_DB.get(product_id)
async def simulate_db_query_all() -> list[dict]:
"""Simulates fetching all products with 500ms latency."""
await asyncio.sleep(0.5)
return list(PRODUCTS_DB.values())
# --- In-Memory Cache Layer ---
memory_cache = TTLCache(maxsize=100, ttl=30)
# --- Redis Client ---
REDIS_URL = os.getenv("REDIS_URL", "")
redis_client: Redis | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global redis_client
if REDIS_URL:
redis_client = Redis.from_url(REDIS_URL, decode_responses=True)
try:
await redis_client.ping()
print("Connected to Redis")
except Exception as e:
print(f"Redis connection failed: {e}. Running without Redis.")
redis_client = None
else:
print("No REDIS_URL set. Running with in-memory cache only.")
yield
if redis_client:
await redis_client.close()
app = FastAPI(title="Product Catalog API - Caching Demo", lifespan=lifespan)
@app.get("/products/{product_id}")
async def get_product(product_id: str):
start = time.perf_counter()
cache_key = f"product:{product_id}"
# Layer 1: Check in-memory cache
if cache_key in memory_cache:
elapsed = time.perf_counter() - start
return {
"data": memory_cache[cache_key],
"cache": "memory",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 2: Database (cache miss)
product = await simulate_db_query(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
memory_cache[cache_key] = product
elapsed = time.perf_counter() - start
return {
"data": product,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}
@app.get("/products")
async def get_all_products():
start = time.perf_counter()
cache_key = "products:all"
# Layer 1: Check in-memory cache
if cache_key in memory_cache:
elapsed = time.perf_counter() - start
return {
"data": memory_cache[cache_key],
"cache": "memory",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 2: Database (cache miss)
products = await simulate_db_query_all()
memory_cache[cache_key] = products
elapsed = time.perf_counter() - start
return {
"data": products,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}Test caching behavior
With both endpoints using cache-aside, you can now observe the dramatic difference between a cache miss and a cache hit.
- Open your browser and visit http://127.0.0.1:8000/products/1.
- Look at the latency_ms and cache fields in the JSON response.
The first request should show "cache": "miss" with a latency around 200ms.
- Refresh the same page immediately.
The second request should show "cache": "memory" with a latency under 1ms. That is the cache-aside pattern in action.
- Wait 30 seconds, then refresh again.
After the TTL expires, the cache entry is gone. You should see "cache": "miss" and 200ms latency again, proving that expiration works correctly.
- Test the list endpoint at http://127.0.0.1:8000/products with the same pattern: first request is a miss (~500ms), second request is a memory hit (<1ms).
Still seeing 200ms on the second request?
Check that you saved main.py after updating both endpoints. The dev server must reload before changes take effect.
Also verify that the memory_cache variable is defined at module level (outside any function). If it is inside a function, it gets recreated on every call.
help me debug why my cache-aside pattern is not reducing latency
Your API now serves repeated requests in under 1ms instead of 200ms. Next up, you will connect Redis so your cache persists across server restarts and can be shared between multiple workers.
Connect Redis via Upstash
Your in-memory cache is fast, but it has a limitation. If you restart the server or run multiple workers, the cache is lost.
In this step, you'll add Redis as a second caching layer that lives outside your Python process. Redis persists data across restarts and can be shared between workers. You'll connect to a free hosted Redis database on Upstash and implement a two-layer cache: memory first, then Redis, then the database.
In this step, get ready to:
- Create a free Upstash Redis database and get the connection URL.
- Set the REDIS_URL environment variable and verify the connection.
- Implement two-layer caching (memory + Redis) in both endpoints.
Create a free Upstash Redis database
Upstash provides a hosted Redis service with a generous free tier (256 MB storage, 500K commands per month). You'll create a database and grab its connection URL.
- Open https://console.upstash.com in your browser and sign up for a free account (or log in if you already have one).
- From the Redis tab, click + Create Database in the upper right corner.
- Enter a name for your database (e.g. product-cache).
- For Primary Region, select the region closest to you (e.g. us-east-1).
- Click Next, select the Free plan, and your database is created.
Once your database is ready, you need the connection URL in rediss:// format.
- On your database page, find the connection details section. You'll see the Endpoint, Port, and Password (also called Token).
- Look for a connection string in the format rediss://default:YOUR_PASSWORD@YOUR_ENDPOINT:YOUR_PORT. Copy this URL. You may find it in a connection example or you can assemble it from the individual fields.
Why rediss:// and not redis://?
The extra s in rediss:// means the connection uses TLS encryption. Upstash enables TLS by default on all databases, so your data is encrypted in transit.
Keep your URL safe
Your connection URL contains your database password. Never commit it to version control. You'll pass it as an environment variable instead.
Set the REDIS_URL environment variable and start the server
Your code already reads REDIS_URL from the environment using os.getenv("REDIS_URL", ""). The lifespan function connects to Redis when this variable is set. You just need to provide the value.
What does the lifespan code do?
Look at the lifespan function already in your main.py. When REDIS_URL is set, it creates a Redis client with Redis.from_url() and pings the server to verify the connection.
If the ping succeeds, it prints Connected to Redis. If it fails, the app gracefully falls back to memory-only caching. When the app shuts down (after yield), the client is closed cleanly.
- Stop your running server if it's still active (press Ctrl+C).
- Start the server with your REDIS_URL by running this command (paste your own URL from Upstash):
REDIS_URL="rediss://default:your-password-here@your-endpoint:your-port" fastapi dev main.py- Check your terminal output. You should see Connected to Redis printed before the Uvicorn startup message.
Seeing 'Redis connection failed'?
Double-check that your REDIS_URL starts with rediss:// (two s's for TLS). Make sure you copied the full password and endpoint from Upstash without extra spaces.
If you see No REDIS_URL set instead, make sure you put the variable on the same line before the fastapi dev command with no line break between them.
help me debug my Redis connection to Upstash
Implement two-layer caching in both endpoints
Right now your endpoints only check the in-memory cache before hitting the database. You'll insert a Redis layer between them. The lookup order becomes: memory (fastest) then Redis (fast, shared) then database (slow).
What is two-layer caching?
Memory is the fastest cache but disappears on restart. Redis is slightly slower (network hop) but persists and is shared across processes. By checking memory first and Redis second, you get the speed of local memory with the durability of Redis as a fallback.
On a Redis hit, you also backfill the memory cache. This means the next request for the same key skips the network call entirely.
- In main.py, find the get_product function. Locate the comment that says:
# Layer 2: Database (cache miss)- Replace everything from that comment to the end of the function with the following code:
# Layer 2: Check Redis cache
if redis_client:
cached = await redis_client.get(cache_key)
if cached:
product = json.loads(cached)
memory_cache[cache_key] = product
elapsed = time.perf_counter() - start
return {
"data": product,
"cache": "redis",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 3: Database (cache miss)
product = await simulate_db_query(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# Store in both caches
memory_cache[cache_key] = product
if redis_client:
await redis_client.setex(cache_key, 60, json.dumps(product))
elapsed = time.perf_counter() - start
return {
"data": product,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}What does this code do?
- Layer 2 checks Redis using redis_client.get(cache_key). If found, it deserializes the JSON and backfills memory_cache so the next request is even faster.
- Layer 3 queries the database on a full miss. After getting the result, it stores in both memory_cache and Redis (using setex with a 60-second TTL).
- The response's cache field tells you which layer served the request: memory, redis, or miss.
- Now find the get_all_products function. Locate the same comment pattern:
# Layer 2: Database (cache miss)- Replace everything from that comment to the end of the function with:
# Layer 2: Check Redis cache
if redis_client:
cached = await redis_client.get(cache_key)
if cached:
products = json.loads(cached)
memory_cache[cache_key] = products
elapsed = time.perf_counter() - start
return {
"data": products,
"cache": "redis",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 3: Database (cache miss)
products = await simulate_db_query_all()
# Store in both caches
memory_cache[cache_key] = products
if redis_client:
await redis_client.setex(cache_key, 60, json.dumps(products))
elapsed = time.perf_counter() - start
return {
"data": products,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}- Save main.py. Your server should auto-reload.
- Open http://127.0.0.1:8000/docs in your browser and test the GET /products/1 endpoint three times:
- First request: you should see "cache": "miss" with a latency around 200ms.
- Second request: you should see "cache": "memory" with latency under 1ms.
- Wait 30 seconds (for the memory cache TTL to expire), then request again: you should see "cache": "redis" with latency of a few milliseconds. The data was still in Redis even though the memory cache expired.
Why is the Redis TTL longer than the memory TTL?
Your memory cache expires after 30 seconds (TTLCache(ttl=30)) but Redis entries last 60 seconds (setex(..., 60, ...)). This intentional gap means Redis acts as a safety net. When the fast memory cache expires, Redis catches the request before it hits the slow database.
Still seeing only 'memory' and 'miss'?
Make sure Connected to Redis printed in your terminal when the server started. If it says No REDIS_URL set, you need to restart the server with the REDIS_URL variable on the same line as the command.
Also confirm your code checks if redis_client: (not if redis_client is not None:). Both work, but make sure redis_client was not set to None by a failed connection attempt.
help me debug why my Redis cache layer isn't being used
✔️ Awesome, I've got everything!
Great. Double check you've saved main.py and that your server is showing Connected to Redis in the terminal.
ⓧ I'd like to double check the full code
import asyncio
import json
import os
import time
from cachetools import TTLCache
from fastapi import FastAPI, HTTPException
from redis.asyncio import Redis
from contextlib import asynccontextmanager
# --- Simulated Database ---
PRODUCTS_DB = {
"1": {"id": "1", "name": "Wireless Headphones", "price": 79.99, "category": "electronics", "stock": 150},
"2": {"id": "2", "name": "Running Shoes", "price": 129.99, "category": "sports", "stock": 75},
"3": {"id": "3", "name": "Coffee Maker", "price": 49.99, "category": "kitchen", "stock": 200},
"4": {"id": "4", "name": "Yoga Mat", "price": 29.99, "category": "sports", "stock": 300},
"5": {"id": "5", "name": "Bluetooth Speaker", "price": 59.99, "category": "electronics", "stock": 90},
}
async def simulate_db_query(product_id: str) -> dict | None:
"""Simulates a slow database query with 200ms latency."""
await asyncio.sleep(0.2)
return PRODUCTS_DB.get(product_id)
async def simulate_db_query_all() -> list[dict]:
"""Simulates fetching all products with 500ms latency."""
await asyncio.sleep(0.5)
return list(PRODUCTS_DB.values())
# --- In-Memory Cache Layer ---
memory_cache = TTLCache(maxsize=100, ttl=30)
# --- Redis Client ---
REDIS_URL = os.getenv("REDIS_URL", "")
redis_client: Redis | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global redis_client
if REDIS_URL:
redis_client = Redis.from_url(REDIS_URL, decode_responses=True)
try:
await redis_client.ping()
print("Connected to Redis")
except Exception as e:
print(f"Redis connection failed: {e}. Running without Redis.")
redis_client = None
else:
print("No REDIS_URL set. Running with in-memory cache only.")
yield
if redis_client:
await redis_client.close()
app = FastAPI(title="Product Catalog API - Caching Demo", lifespan=lifespan)
@app.get("/products/{product_id}")
async def get_product(product_id: str):
start = time.perf_counter()
cache_key = f"product:{product_id}"
# Layer 1: Check in-memory cache
if cache_key in memory_cache:
elapsed = time.perf_counter() - start
return {
"data": memory_cache[cache_key],
"cache": "memory",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 2: Check Redis cache
if redis_client:
cached = await redis_client.get(cache_key)
if cached:
product = json.loads(cached)
memory_cache[cache_key] = product
elapsed = time.perf_counter() - start
return {
"data": product,
"cache": "redis",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 3: Database (cache miss)
product = await simulate_db_query(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# Store in both caches
memory_cache[cache_key] = product
if redis_client:
await redis_client.setex(cache_key, 60, json.dumps(product))
elapsed = time.perf_counter() - start
return {
"data": product,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}
@app.get("/products")
async def get_all_products():
start = time.perf_counter()
cache_key = "products:all"
# Layer 1: Check in-memory cache
if cache_key in memory_cache:
elapsed = time.perf_counter() - start
return {
"data": memory_cache[cache_key],
"cache": "memory",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 2: Check Redis cache
if redis_client:
cached = await redis_client.get(cache_key)
if cached:
products = json.loads(cached)
memory_cache[cache_key] = products
elapsed = time.perf_counter() - start
return {
"data": products,
"cache": "redis",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 3: Database (cache miss)
products = await simulate_db_query_all()
# Store in both caches
memory_cache[cache_key] = products
if redis_client:
await redis_client.setex(cache_key, 60, json.dumps(products))
elapsed = time.perf_counter() - start
return {
"data": products,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}You now have a two-layer caching system. Memory catches the hottest requests in microseconds. Redis catches everything else in a few milliseconds. Only truly cold requests hit the 200ms database. Next up, you'll implement cache invalidation so updates don't serve stale data.
Implement Cache Invalidation
In the previous step, you connected your API to Redis via Upstash and built a two-layer cache that dramatically speeds up reads.
But what happens when data changes? If you update a product's price, anyone reading from cache still sees the old price until the TTL expires. That window of stale data is unacceptable in production. You need cache invalidation: the ability to proactively refresh or remove cached entries when the source data changes.
In this step, get ready to:
- Create a write-through update endpoint that keeps caches in sync with the database.
- Add cache management endpoints for manual clearing and monitoring.
- Test the full read-write-read lifecycle to confirm data stays fresh.
Create the update endpoint with write-through invalidation
Write-through caching means that every time you write to the database, you immediately update all caches with the fresh data. This ensures readers never see stale values after a write completes.
- In main.py, scroll to the bottom of the file (after the get_all_products function) and add the following PUT endpoint:
@app.put("/products/{product_id}")
async def update_product(product_id: str, price: float | None = None, stock: int | None = None):
"""Update a product with write-through cache invalidation."""
if product_id not in PRODUCTS_DB:
raise HTTPException(status_code=404, detail="Product not found")
# Update the "database"
if price is not None:
PRODUCTS_DB[product_id]["price"] = price
if stock is not None:
PRODUCTS_DB[product_id]["stock"] = stock
updated_product = PRODUCTS_DB[product_id]
# Write-through: update caches with fresh data
cache_key = f"product:{product_id}"
memory_cache[cache_key] = updated_product
memory_cache.pop("products:all", None)
if redis_client:
await redis_client.setex(cache_key, 60, json.dumps(updated_product))
await redis_client.delete("products:all")
return {"data": updated_product, "message": "Product updated, caches refreshed"}What does this code do?
This endpoint updates a product's price or stock, then immediately refreshes every cache layer so no stale data lingers.
- memory_cache[cache_key] = updated_product overwrites the single-product entry in memory with the fresh data.
- memory_cache.pop("products:all", None) removes the cached list of all products, since it now contains outdated information.
- redis_client.setex writes the updated product into Redis with a fresh 60-second TTL.
- redis_client.delete("products:all") removes the list cache from Redis so the next list request rebuilds it from the database.
- Save main.py.
- Your dev server should auto-reload. Open http://127.0.0.1:8000/docs in your browser to see the new PUT /products/{product_id} endpoint listed in the interactive docs.
Don't see the PUT endpoint in /docs?
- Make sure you saved main.py and check the terminal for error messages. A syntax error will prevent the reload.
- Confirm the new function is outside and below the get_all_products function, not accidentally indented inside it.
Why isn't my PUT endpoint appearing in the FastAPI docs?
Add cache management endpoints
Sometimes you need to clear all caches manually, or inspect what is currently cached. These utility endpoints give you visibility and control over your caching layers.
- Below the update_product function, add the cache-clearing endpoint:
@app.delete("/cache")
async def clear_all_caches():
"""Manually clear all caches."""
memory_cache.clear()
if redis_client:
keys = []
async for key in redis_client.scan_iter("product*"):
keys.append(key)
if keys:
await redis_client.delete(*keys)
return {"message": "All caches cleared"}What does this code do?
memory_cache.clear() wipes every entry from the in-memory TTLCache in one call.
For Redis, you cannot just call a blanket flush (that would erase all keys in the database, including ones from other applications). Instead, scan_iter("product*") iterates over only the keys your app created, and redis_client.delete(*keys) removes them in a single batch call.
- Below the clear_all_caches function, add the cache stats endpoint:
@app.get("/cache/stats")
async def cache_stats():
"""View current cache state."""
stats = {
"memory_cache": {
"items": len(memory_cache),
"maxsize": memory_cache.maxsize,
"ttl_seconds": int(memory_cache.ttl),
}
}
if redis_client:
keys = []
async for key in redis_client.scan_iter("product*"):
keys.append(key)
stats["redis_cache"] = {
"keys": keys,
"count": len(keys),
}
return statsWhat does this code do?
This endpoint reports the current state of both cache layers. It shows how many items are in memory, the configured max size, the TTL, and which keys currently exist in Redis.
This is your observability tool. In production systems, cache monitoring helps you understand hit ratios, identify bloated caches, and verify that invalidation is working correctly.
- Save main.py.
- Refresh http://127.0.0.1:8000/docs in your browser. You should now see DELETE /cache and GET /cache/stats listed alongside your other endpoints.
Seeing a server error on reload?
- Check for indentation issues. Each new endpoint function should start at the top level (no indentation), not nested inside another function.
- Verify that async for key in redis_client.scan_iter("product*"): uses async for, not just for. The Redis client returns an async iterator.
Help me debug my cache management endpoints in FastAPI.
Test the full cache lifecycle
Now you will run through the complete read-write-read cycle to confirm that cache invalidation keeps data fresh.
- Open http://127.0.0.1:8000/docs in your browser.
- Click on GET /products/1, then click Try it out, enter 1 for the product_id, and click Execute.
- Note the response shows "cache": "miss" and "latency_ms" around 200ms.
- Click Execute again. This time you should see "cache": "memory" with latency under 1ms.
- Now click on PUT /products/{product_id}, click Try it out, enter 1 for product_id and 99.99 for price, then click Execute.
- Go back to GET /products/1 and click Execute again. The response should show "price": 99.99 (fresh data from cache, not the old 79.99).
Why is the data fresh and not stale?
When you called PUT, the write-through logic immediately updated both the memory cache and Redis with the new price. The next GET finds the updated value in cache without ever hitting the simulated database again.
Without write-through invalidation, you would have seen the old price (79.99) until the TTL expired, up to 30 seconds for the memory cache or 60 seconds for Redis.
- Test the stats endpoint by clicking on GET /cache/stats and clicking Execute. You should see the number of items currently cached in memory and the Redis key names.
Still seeing the old price after updating?
- Make sure your PUT endpoint code includes both the memory_cache[cache_key] = updated_product line and the redis_client.setex call. Missing either one means that cache layer still holds stale data.
- Try clicking DELETE /cache to clear everything, then re-read. If the fresh price appears after clearing, your invalidation logic has a typo.
Why am I seeing stale data after updating a product?
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py and that all three new endpoints appear in your /docs page.
ⓧ I'd like to double check the full code
Here is the complete main.py at this point in the project:
import asyncio
import json
import os
import time
from cachetools import TTLCache
from fastapi import FastAPI, HTTPException
from redis.asyncio import Redis
from contextlib import asynccontextmanager
# --- Simulated Database ---
PRODUCTS_DB = {
"1": {"id": "1", "name": "Wireless Headphones", "price": 79.99, "category": "electronics", "stock": 150},
"2": {"id": "2", "name": "Running Shoes", "price": 129.99, "category": "sports", "stock": 75},
"3": {"id": "3", "name": "Coffee Maker", "price": 49.99, "category": "kitchen", "stock": 200},
"4": {"id": "4", "name": "Yoga Mat", "price": 29.99, "category": "sports", "stock": 300},
"5": {"id": "5", "name": "Bluetooth Speaker", "price": 59.99, "category": "electronics", "stock": 90},
}
async def simulate_db_query(product_id: str) -> dict | None:
"""Simulates a slow database query with 200ms latency."""
await asyncio.sleep(0.2)
return PRODUCTS_DB.get(product_id)
async def simulate_db_query_all() -> list[dict]:
"""Simulates fetching all products with 500ms latency."""
await asyncio.sleep(0.5)
return list(PRODUCTS_DB.values())
# --- In-Memory Cache Layer ---
memory_cache = TTLCache(maxsize=100, ttl=30)
# --- Redis Client ---
REDIS_URL = os.getenv("REDIS_URL", "")
redis_client: Redis | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global redis_client
if REDIS_URL:
redis_client = Redis.from_url(REDIS_URL, decode_responses=True)
try:
await redis_client.ping()
print("Connected to Redis")
except Exception as e:
print(f"Redis connection failed: {e}. Running without Redis.")
redis_client = None
else:
print("No REDIS_URL set. Running with in-memory cache only.")
yield
if redis_client:
await redis_client.close()
app = FastAPI(title="Product Catalog API - Caching Demo", lifespan=lifespan)
@app.get("/products/{product_id}")
async def get_product(product_id: str):
start = time.perf_counter()
cache_key = f"product:{product_id}"
# Layer 1: Check in-memory cache
if cache_key in memory_cache:
elapsed = time.perf_counter() - start
return {
"data": memory_cache[cache_key],
"cache": "memory",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 2: Check Redis cache
if redis_client:
cached = await redis_client.get(cache_key)
if cached:
product = json.loads(cached)
memory_cache[cache_key] = product
elapsed = time.perf_counter() - start
return {
"data": product,
"cache": "redis",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 3: Database (cache miss)
product = await simulate_db_query(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# Store in both caches
memory_cache[cache_key] = product
if redis_client:
await redis_client.setex(cache_key, 60, json.dumps(product))
elapsed = time.perf_counter() - start
return {
"data": product,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}
@app.get("/products")
async def get_all_products():
start = time.perf_counter()
cache_key = "products:all"
# Layer 1: Check in-memory cache
if cache_key in memory_cache:
elapsed = time.perf_counter() - start
return {
"data": memory_cache[cache_key],
"cache": "memory",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 2: Check Redis cache
if redis_client:
cached = await redis_client.get(cache_key)
if cached:
products = json.loads(cached)
memory_cache[cache_key] = products
elapsed = time.perf_counter() - start
return {
"data": products,
"cache": "redis",
"latency_ms": round(elapsed * 1000, 2),
}
# Layer 3: Database (cache miss)
products = await simulate_db_query_all()
# Store in both caches
memory_cache[cache_key] = products
if redis_client:
await redis_client.setex(cache_key, 60, json.dumps(products))
elapsed = time.perf_counter() - start
return {
"data": products,
"cache": "miss",
"latency_ms": round(elapsed * 1000, 2),
}
@app.put("/products/{product_id}")
async def update_product(product_id: str, price: float | None = None, stock: int | None = None):
"""Update a product with write-through cache invalidation."""
if product_id not in PRODUCTS_DB:
raise HTTPException(status_code=404, detail="Product not found")
# Update the "database"
if price is not None:
PRODUCTS_DB[product_id]["price"] = price
if stock is not None:
PRODUCTS_DB[product_id]["stock"] = stock
updated_product = PRODUCTS_DB[product_id]
# Write-through: update caches with fresh data
cache_key = f"product:{product_id}"
memory_cache[cache_key] = updated_product
memory_cache.pop("products:all", None)
if redis_client:
await redis_client.setex(cache_key, 60, json.dumps(updated_product))
await redis_client.delete("products:all")
return {"data": updated_product, "message": "Product updated, caches refreshed"}
@app.delete("/cache")
async def clear_all_caches():
"""Manually clear all caches."""
memory_cache.clear()
if redis_client:
keys = []
async for key in redis_client.scan_iter("product*"):
keys.append(key)
if keys:
await redis_client.delete(*keys)
return {"message": "All caches cleared"}
@app.get("/cache/stats")
async def cache_stats():
"""View current cache state."""
stats = {
"memory_cache": {
"items": len(memory_cache),
"maxsize": memory_cache.maxsize,
"ttl_seconds": int(memory_cache.ttl),
}
}
if redis_client:
keys = []
async for key in redis_client.scan_iter("product*"):
keys.append(key)
stats["redis_cache"] = {
"keys": keys,
"count": len(keys),
}
return statsSecret mission
Cache Stampede Protection
What happens when a popular cache key expires and 20 requests all try to rebuild it at the same time? Your database gets hammered with redundant queries. Add distributed locking to ensure only one request rebuilds the cache while others wait patiently.
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 uses the Upstash free tier and local files, so there are no ongoing costs.
Resources you used:
- Your local project folder containing main.py and the Python virtual environment.
- Upstash Redis database (free tier).
- Upstash account.
Keep everything running
No action needed. Choose this if you're still actively building or experimenting with caching patterns.
- Your project folder stays on your Desktop ready to use.
- Your Upstash Redis database remains active on the free tier at no cost.
- Free tier databases are archived after 14 days of inactivity. Upstash sends warning emails before archival, and no data is lost.
Pause - I'll come back to this later
Stop running processes to free up memory. Your files and Upstash database stay intact.
- Stop the FastAPI dev server by pressing Ctrl+C in the terminal where it is running.
- Your Upstash database will remain on the free tier at no cost. If left unused for 14+ days, Upstash archives it with a backup you can restore later.
Delete - I don't want to use this again
Remove all project resources and start fresh.
- Stop the FastAPI dev server by pressing Ctrl+C in the terminal.
- Delete your Upstash Redis database by logging in to the Upstash Console, selecting your database from the Redis tab, scrolling to the bottom of the database details page, and clicking Delete Database.
- Delete your local project folder. On macOS/Linux, run:
cd ~/Desktop
rm -rf fastapi-caching- On Windows, run:
cd ~\Desktop
Remove-Item -Recurse -Force fastapi-caching- If you want to delete your Upstash account entirely, go to Account > Settings > Delete Account in the Upstash Console. You must delete all databases first.
That's a Wrap!
That's a Wrap!
Nice work! You've built a production-grade two-layer caching system from scratch, going from 200ms+ response times all the way down to sub-2ms cache hits.
You now understand exactly what happens behind the scenes when a cache-aside lookup fires, when Redis backfills memory, and when write-through invalidation keeps data fresh.
You've learned how to:
- Build a FastAPI API with a two-layer cache architecture (in-memory TTLCache for hot-path speed, Redis via Upstash for distributed persistence).
- Implement the cache-aside pattern with latency timing and cache-hit metadata baked into every response.
- Create write-through cache invalidation on updates and build monitoring endpoints to observe cache state in real time.
- Secret Mission: Add cache stampede protection using Redis distributed locking to handle thundering herd scenarios.
Ready to quiz yourself?