How to Add Redis to a FastAPI App

FastAPI handles async endpoints well, and Redis is usually the second piece of infrastructure a FastAPI service gets. The code is short, and the time goes into client lifecycle, key expiry, a Redis server for every laptop and a private ElastiCache or Memorystore instance.

This guide adds Redis to a FastAPI app with Encore, redis-py's asyncio client and fastapi-cache2. Each version builds a read-through cache, a per-key rate limit and a sorted-set leaderboard.

What you need to decide:

• Who owns the Redis client and its connection pool.
• How keys expire, so the cache cannot grow without bound.
• Where Redis comes from for local development and tests.
• Who creates and secures the production instance.

Redis options for FastAPI at a glance

Encore: Redis for your existing FastAPI app

Encore combines the Redis database your FastAPI code talks to with the server behind it, locally and in your cloud, so the same declaration gives you a local Redis under encore run and a managed instance after deploy. It is an infrastructure SDK you add to an existing app, and FastAPI keeps its routing, dependencies and Pydantic models.

The project needs encore-dev next to FastAPI and Uvicorn in pyproject.toml, and Python 3.9 or later:

[project]
name = "shop"
version = "0.1.0"
requires-python = ">=3.9"
dependencies = ["fastapi", "uvicorn", "encore-dev"]

An encore.service.json in the project root names the service and exposes its HTTP routes:

{
  "name": "shop",
  "lang": "python",
  "api": [
    { "type": "http", "prefix": "/", "name": "shop", "expose": true }
  ]
}

The Redis database is a module-level declaration, and encore.start() runs in FastAPI's lifespan handler so the runtime is ready before the first request. The cache endpoint reads the key, falls back to the slow lookup on a miss and writes the result back with a 60 second expiry:

import asyncio
import json
import os
from contextlib import asynccontextmanager
from typing import Annotated

import encore
import uvicorn
from encore.storage import redis
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

# Encore provisions this Redis database locally under `encore run`
# and in your cloud account when the app is deployed.
cache = redis.Database("cache")


@asynccontextmanager
async def lifespan(app: FastAPI):
    await encore.start()
    yield


app = FastAPI(lifespan=lifespan)


async def load_product(product_id: int) -> dict:
    # Stand-in for a slow database query or upstream API call.
    await asyncio.sleep(0.5)
    return {"id": product_id, "name": f"Product {product_id}", "price_cents": 1999}


@app.get("/products/{product_id}")
async def get_product(product_id: int) -> dict:
    key = f"product:{product_id}"
    cached = await cache.string.get[str](key)
    if cached is not None:
        return {"source": "cache", "product": json.loads(cached)}

    product = await load_product(product_id)
    await cache.string.set(key, json.dumps(product), ex=60)
    return {"source": "origin", "product": product}

The rate limit keeps one counter per API key. The counter is created with its expiry before it is incremented, which means a crash between the two calls can never leave a key that lives forever:

LIMIT = 5
WINDOW_SECONDS = 60


@app.post("/search")
async def search(x_api_key: Annotated[str, Header()]) -> dict:
    key = f"ratelimit:search:{x_api_key}"
    # Create the counter with its expiry first, so it can never be left without one.
    await cache.string.set(key, 0, ex=WINDOW_SECONDS, nx=True)
    count = await cache.string.incr(key)
    if count > LIMIT:
        raise HTTPException(status_code=429, detail="too many requests")
    return {"ok": True, "used": count, "limit": LIMIT}

The leaderboard uses a sorted set, where zincrby adds points to a player and zrevrange returns the top ten with their scores:

class Score(BaseModel):
    player: str
    points: int


@app.post("/scores")
async def add_score(score: Score) -> dict:
    total = await cache.sorted_set.zincrby("leaderboard", score.points, score.player)
    return {"player": score.player, "total": total}


@app.get("/leaderboard")
async def leaderboard() -> list[dict]:
    top = await cache.sorted_set.zrevrange[str]("leaderboard", 0, 9, withscores=True)
    return [{"player": player, "points": points} for player, points in top]


if __name__ == "__main__":
    host, _, port = os.environ.get("ENCORE_LISTEN_ADDR", "0.0.0.0:8000").rpartition(":")
    uvicorn.run(app, host=host or "0.0.0.0", port=int(port))

encore run starts Redis locally along with the app, so the first request to /products/7 takes the half second of the simulated lookup and the second comes back from the cache in a few milliseconds:

The sixth call to /search with the same key inside a minute returns 429, and the local development dashboard lists each request with its timing, which makes a cache that never hits easy to spot. On deploy, Encore provisions ElastiCache on AWS or Memorystore on GCP in your own cloud account, inside a private network that only the services can reach.

Because the cache is declared in the same file as the endpoint, a coding agent that adds caching to a route writes the declaration and the code together and checks the hit locally. Cloud access sits with Encore's IAM role, not the agent.

Good to know

Keys are plain strings, so keep a prefix per use (product:, ratelimit:) to make them easy to scan and expire. The typed accessors such as get[str] decode the stored bytes, and calls without a type return bytes.

Try Encore →

redis-py asyncio

redis-py is the official Python client, and its asyncio module is the right one for async def endpoints. Create the client once in the lifespan and close it on shutdown, because a client per request opens a new pool each time:

import os
from contextlib import asynccontextmanager

import redis.asyncio as redis
from fastapi import FastAPI, Request


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = redis.from_url(
        os.environ.get("REDIS_URL", "redis://localhost:6379/0"),
        decode_responses=True,
    )
    yield
    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)


@app.get("/products/{product_id}")
async def get_product(product_id: int, request: Request) -> dict:
    r = request.app.state.redis
    key = f"product:{product_id}"
    if (cached := await r.get(key)) is not None:
        return {"source": "cache", "product": cached}
    product = f"Product {product_id}"
    await r.set(key, product, ex=60)
    return {"source": "origin", "product": product}

A rate limit with redis-py looks the same as above, and a pipeline with transaction=True can group the SET NX EX and INCR into one round trip.

Good to know

You run Redis yourself for development and CI, usually as a container or a system service, and REDIS_URL has to be set per environment. In production that URL points at an ElastiCache or Memorystore endpoint that only the application's network can reach.

fastapi-cache2

fastapi-cache2 caches whole endpoint responses through a @cache decorator, with Redis as one of its backends. It suits read-heavy endpoints whose response depends only on the path and query:

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
from redis import asyncio as aioredis


@asynccontextmanager
async def lifespan(app: FastAPI):
    client = aioredis.from_url("redis://localhost:6379")
    FastAPICache.init(RedisBackend(client), prefix="api")
    yield


app = FastAPI(lifespan=lifespan)


@app.get("/products/{product_id}")
@cache(expire=60)
async def get_product(product_id: int) -> dict:
    return {"id": product_id, "name": f"Product {product_id}"}

Good to know

Response caching covers one use case, so rate limits, leaderboards and queues still need a regular client next to it. Responses that depend on the current user need a custom key builder, or one user's data can be served to another.

Redis in production

On AWS, ElastiCache runs inside a VPC, and the ElastiCache documentation notes that clusters are accessed from EC2 instances or other resources in the same VPC rather than from the internet. That means subnet groups, a security group that admits only the service, and a way for the service to learn the endpoint. Set maxmemory-policy deliberately, since a cache that evicts nothing will start failing writes when memory runs out. Redis changed its license in 2024 and the Linux Foundation now hosts the Valkey fork, which ElastiCache and Memorystore both offer, and the client code above works against either. With Encore, the instance, network and access rules are created in your account from the declaration in the code.

How to choose

• Encore if you want Redis running locally and in your AWS or GCP account from the same declaration.
• redis-py asyncio if you already run Redis and want direct access to every command.
• fastapi-cache2 if all you need is response caching on a few read-heavy endpoints.

Python support in Encore is in beta, and you can sign up for access to try it with your FastAPI app.

Related guides

• How to add Pub/Sub to a FastAPI app
• How to deploy FastAPI to Google Cloud Run
• How to deploy FastAPI to AWS
• FastAPI vs Django vs Flask
• Best Python backend frameworks
• How to add Redis to an Axum app