FastAPI + Redis: Basic Application of Caching and State Management

In web development, we often face two core challenges: how to quickly respond to requests and how to share temporary state across multiple requests. FastAPI, a high-performance Python web framework, combined with Redis (an in-memory data storage system), effectively addresses these two issues. This article will guide you through understanding how they work together with simple examples.

1. Why FastAPI + Redis?¶

FastAPI: Enables rapid API construction and supports asynchronous request handling, making it suitable for high-concurrency scenarios. However, even with FastAPI’s speed, optimizations are still needed for frequent, repetitive calculations or data queries (e.g., user information, statistics).
Redis: A high-performance in-memory database that supports key-value storage with fast read/write speeds. It also offers features like expiration times and distributed locks, commonly used for caching (reducing redundant computations) and temporary state management (e.g., sessions, counters).

2. Environment Setup¶

First, install the necessary tools:

# 1. Install FastAPI and Uvicorn (web server)
pip install fastapi uvicorn

# 2. Install Redis Python client
pip install redis

Ensure Redis is installed and running locally (default port 6379, no password for direct connection). Verify the connection:

redis-cli  # Enter Redis CLI, execute `ping`; a response of "PONG" confirms success

3. Basic Redis Connection in FastAPI¶

In FastAPI, centralize Redis connections via utility functions to avoid repeated initialization:

from fastapi import FastAPI
import redis
import json

# Initialize Redis connection (simplified here; use connection pools in production)
redis_client = redis.StrictRedis(
    host="localhost",  # Redis server address
    port=6379,         # Default port
    db=0,              # Database number
    decode_responses=True  # Automatically decode responses to strings
)

app = FastAPI()

# Test connection
@app.get("/redis-test")
async def redis_test():
    redis_client.set("test_key", "Hello Redis!")  # Store key-value pair
    return {"message": redis_client.get("test_key")}  # Read key-value pair

Note: decode_responses=True ensures Redis returns strings instead of bytes, simplifying direct processing.

4. Caching: A Tool to Reduce Redundant Computation¶

Caching stores high-frequency, low-update data (e.g., fixed configurations, user profiles) in Redis, allowing subsequent requests to read directly from the cache and avoid redundant computations or database queries.

Example: Simulate a “time-consuming calculation” (e.g., Fibonacci sequence, mimicking complex logic) that is inefficient to recompute on every request. Use Redis to cache results:

# Simulate "time-consuming calculation" (could be database query or complex algorithm)
def slow_calculate(n: int) -> int:
    if n <= 1:
        return n
    return slow_calculate(n-1) + slow_calculate(n-2)

# FastAPI view with caching logic
from fastapi import HTTPException

@app.get("/cache/calculate/{n}")
async def cached_calculate(n: int):
    cache_key = f"fib:{n}"  # Cache key (unique per calculation input)

    # 1. Check cache first
    cached_result = redis_client.get(cache_key)
    if cached_result:
        return {"n": n, "result": int(cached_result), "source": "cache"}

    # 2. Cache miss: perform time-consuming calculation
    try:
        result = slow_calculate(n)
    except Exception as e:
        raise HTTPException(status_code=500, detail="Calculation failed")

    # 3. Store in cache with expiration (10 minutes to avoid permanent storage)
    redis_client.setex(cache_key, 600, str(result))  # setex = set + expiration

    return {"n": n, "result": result, "source": "database"}

Effect: The first request to /cache/calculate/10 triggers computation and caching. Subsequent requests retrieve the result directly from Redis without recomputation.

Redis also stores temporary state (e.g., user sessions, operation counters) to solve FastAPI’s multi-request state sharing problem (in-memory variables are process-bound, while Redis works across processes/servers).

Example: Count user visits (temporary state):

@app.get("/counter/{user_id}")
async def user_counter(user_id: str):
    cache_key = f"counter:{user_id}"  # State key (user-specific)

    # 1. Read current visit count from Redis
    current_count = redis_client.get(cache_key)
    if current_count:
        current_count = int(current_count) + 1  # Increment count
    else:
        current_count = 1  # New user: initial count = 1

    # 2. Update cache with expiration (24 hours, auto-expire for inactive users)
    redis_client.setex(cache_key, 86400, str(current_count))

    return {"user_id": user_id, "visits": current_count}

Effect: A user with user_id=123 sees visits=1 on the first visit and visits=2 on the second. Data persists in Redis even after service restarts.

6. Advanced Tip: Dependency Injection for Code Optimization¶

To avoid repeating Redis connection logic, use FastAPI’s dependency injection to manage the Redis client uniformly:

from fastapi import Depends

# Define Redis connection dependency
async def get_redis_client():
    return redis.StrictRedis(host="localhost", port=6379, db=0, decode_responses=True)

# Use dependency in view functions
@app.get("/example")
async def example(redis_client=Depends(get_redis_client)):
    redis_client.set("key", "value")
    return {"data": redis_client.get("key")}

7. Summary¶

Caching: Store high-frequency, low-update data (e.g., user info, statistics) in Redis to reduce database/computation load.
State Management: Share temporary data across requests using Redis key-value pairs with expiration times (e.g., sessions, counters, permissions).

By combining FastAPI’s asynchronous performance with Redis’s in-memory speed, you can efficiently solve “slow response” and “state sharing” challenges in web development.

Production Notes:
- Configure Redis persistence (to prevent data loss).
- Optimize connection pools (avoid frequent connection creation).
- Use consistent key naming conventions (e.g., fib:{n}, counter:{user_id}) to prevent conflicts.