Redis is an in-memory data store commonly used for caching, rate limiting, and session storage. In a FastAPI application, Redis can help reduce database queries and share data across multiple application workers.
The easiest way to run Redis locally is with Docker:
docker run -d \--name redis \-p 127.0.0.1:6379:6379 \redis:7-alpine
Verify that Redis is running:
docker exec -it redis redis-cli ping
Expected output:
PONG
Redis is now available on localhost:6379.
Install FastAPI, Uvicorn, and the Redis Python client:
pip install fastapi uvicorn redis
The redis package provides an asynchronous client through redis.asyncio, which is suitable for FastAPI’s async endpoints.
Create the following project structure:
fastapi-app/├── app/│ ├── __init__.py│ ├── main.py│ └── redis.py└── requirements.txt
Define the Redis connection in app/redis.py:
import redis.asyncio as redisredis_client = redis.from_url("redis://localhost:6379/0",decode_responses=True,)
The connection URL follows this format:
redis://[username]:[password]@[host]:[port]/[database]
localhost: Redis host.6379: Default Redis port.0: Redis database index.decode_responses=True: Returns strings instead of bytes for text responses.For local development without authentication, the username and password can be omitted.
Use FastAPI’s lifespan handler to initialize and close the Redis client.
# app/main.pyfrom contextlib import asynccontextmanagerfrom fastapi import FastAPI, Requestfrom app.redis import redis_client@asynccontextmanagerasync def lifespan(app: FastAPI):await redis_client.ping()app.state.redis = redis_clienttry:yieldfinally:await redis_client.aclose()app = FastAPI(lifespan=lifespan)@app.get("/health/redis")async def redis_health(request: Request):await request.app.state.redis.ping()return {"status": "Redis is connected"}
The lifespan handler runs before the application starts accepting requests and cleans up the client during shutdown.
The ping() call verifies connectivity. If Redis is unavailable, the application will fail to start rather than silently running without its required dependency.
Add endpoints to store and retrieve values:
from fastapi import HTTPException@app.put("/cache/{key}")async def set_cache(key: str,value: str,request: Request,):client = request.app.state.redisawait client.set(key, value, ex=300)return {"message": "Value stored successfully"}@app.get("/cache/{key}")async def get_cache(key: str, request: Request):client = request.app.state.redisvalue = await client.get(key)if value is None:raise HTTPException(status_code=404,detail="Cache key not found",)return {"key": key, "value": value}
The ex=300 option sets a TTL of five minutes. Redis automatically expires the key after that period.
For a real application, validate keys and values, avoid exposing arbitrary cache access, and use an appropriate serialization format for structured data.
Start the FastAPI development server:
uvicorn app.main:app --reload
Open the interactive API documentation at:
http://127.0.0.1:8000/docs
Try the following endpoints:
| Method | Endpoint | Purpose |
|---|---|---|
| GET | /health/redis | Check the Redis connection |
| PUT | /cache/{key}?value=hello | Store a value |
| GET | /cache/{key} | Retrieve a value |
For example, calling PUT /cache/greeting?value=hello stores the value hello under the key greeting. A subsequent GET request returns that value.