FastAPI State Management: Simple Implementation of Global Variables and Caching

In web application development, state management is a common requirement—for example, when multiple requests need to share temporary data (such as counters or user sessions), or when caching frequently accessed results to improve performance. As a modern Python asynchronous framework, FastAPI provides multiple ways to implement state management, with global variables and caching being the most basic and commonly used methods. This article will explain how to implement these two state management approaches in FastAPI from simple to complex.

一、Global Variables: Simple State Sharing¶

Global variables are the simplest form of state management, where a variable is defined directly in the code and shared/modified by multiple route functions. However, it is crucial to note that FastAPI supports asynchronous operations and may run in a multi-process environment. Thus, the use of global variables requires careful handling of thread safety and multi-process isolation.

1. Global Variables in a Single Process¶

In single-process deployment (e.g., the default Uvicorn single-process startup), global variables can be used directly. Here is a counter example:

from fastapi import FastAPI
import asyncio

app = FastAPI()

# Define a global variable: used to store the counter
count = 0
# Define an asynchronous lock (to solve race conditions when multiple requests modify simultaneously)
lock = asyncio.Lock()

@app.get("/counter")
async def get_counter():
    global count
    async with lock:  # Acquire the lock to ensure only one request modifies 'count' at a time
        count += 1
        return {"current_count": count}

Explanation:
- count is a global variable that increments by 1 with each request.
- asyncio.Lock() is an asynchronous lock that ensures only one request can modify count at a time, preventing data errors caused by concurrent read/write operations (e.g., two requests reading count=1 simultaneously and both incrementing it, resulting in an incorrect final value of 2 instead of 3).
- Each visit to /counter will show the counter incrementing, e.g., 1, 2, 3, etc.

2. Limitations of Global Variables¶

Global variables are only suitable for single-process scenarios and have the following issues:
- Multi-process Isolation: When using Uvicorn --workers N to start multiple processes, each process has its own independent memory space, so global variables cannot be shared across processes (e.g., with N=2, each process’s counter increments independently).
- Memory Dependency: Data stored in global variables is in-memory and will be lost after service restart; it does not support persistence.
- Thread Safety Risks: Without proper locking, multiple requests in an asynchronous environment may modify the global variable simultaneously, leading to data errors.

二、Caching: More Efficient State Management¶

When dealing with frequently accessed data (e.g., user information, configuration parameters) or data shared across users, global variables are inefficient. In such cases, caching should be used. Caching is an “intermediate layer” for temporary data storage, avoiding redundant calculations or database queries. Common implementations include in-memory caching, professional caching libraries (e.g., cachetools), and distributed caching (e.g., Redis).

1. In-Memory Caching: Simulating with a Dictionary¶

The simplest caching method is to directly use a Python dictionary to store key-value pairs, suitable for small-scale data and development environments:

from fastapi import FastAPI
import time

app = FastAPI()

# Simulate a database query (replace with actual database operations in production)
def get_user_from_db(user_id: int):
    time.sleep(1)  # Simulate 1-second query delay
    return {"user_id": user_id, "name": f"User_{user_id}"}

# Define an in-memory cache dictionary
user_cache = {}

@app.get("/user/{user_id}")
async def get_user(user_id: int):
    # 1. Check the cache first
    if user_id in user_cache:
        return {"source": "cache", "data": user_cache[user_id]}

    # 2. If cache miss, query the database and cache the result
    user_data = get_user_from_db(user_id)
    user_cache[user_id] = user_data  # Store in cache
    return {"source": "database", "data": user_data}

Effect:
- On the first request to /user/1, it takes 1 second (simulating a database query) and caches the result.
- On subsequent requests to /user/1, the result is retrieved directly from the cache, avoiding waiting and significantly improving performance.

2. Using cachetools for Automatic Caching¶

cachetools is a Python library that provides various caching strategies (e.g., LRU, TTL), simplifying manual dictionary management:

from fastapi import FastAPI
from cachetools import LRUCache, TTLCache
from typing import Optional

app = FastAPI()

# Method 1: LRU Cache (stores up to 10 entries, evicting the least recently used)
user_cache_lru = LRUCache(maxsize=10)

# Method 2: TTL Cache (entries expire after 10 seconds)
user_cache_ttl = TTLCache(maxsize=10, ttl=10)

# Simulate user query (same as previous example)
def get_user_from_db(user_id: int):
    time.sleep(1)
    return {"user_id": user_id, "name": f"User_{user_id}"}

@app.get("/user/lru/{user_id}")
async def get_user_lru(user_id: int):
    if user_id in user_cache_lru:
        return {"source": "lru_cache", "data": user_cache_lru[user_id]}
    user_data = get_user_from_db(user_id)
    user_cache_lru[user_id] = user_data
    return {"source": "database", "data": user_data}

@app.get("/user/ttl/{user_id}")
async def get_user_ttl(user_id: int):
    if user_id in user_cache_ttl:
        return {"source": "ttl_cache", "data": user_cache_ttl[user_id]}
    user_data = get_user_from_db(user_id)
    user_cache_ttl[user_id] = user_data
    return {"source": "database", "data": user_data}

Key Features:
- LRUCache(maxsize=10): Caches up to 10 entries, automatically evicting the least recently used key when the limit is exceeded.
- TTLCache(maxsize=10, ttl=10): Entries expire after 10 seconds, preventing permanent memory occupation by stale data.

3. Distributed Caching: Redis Integration (Advanced)¶

For cross-server shared caching (e.g., multi-instance deployment) or persistent data, a professional caching service like Redis is recommended. Here’s an example of integrating Redis with FastAPI:

from fastapi import FastAPI
import redis.asyncio as redis
import json

app = FastAPI()

# Connect to Redis (install redis first: pip install redis)
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

@app.get("/user/redis/{user_id}")
async def get_user_redis(user_id: int):
    # 1. Try to read from Redis cache
    cached_data = await redis_client.get(f"user:{user_id}")
    if cached_data:
        return {"source": "redis_cache", "data": json.loads(cached_data)}

    # 2. Cache miss: query the database
    user_data = get_user_from_db(user_id)
    # 3. Store in Redis with a 10-minute expiration
    await redis_client.setex(
        f"user:{user_id}",  # Key: user:1
        600,                # Expiration: 600 seconds (10 minutes)
        json.dumps(user_data)  # Value: JSON-serialized data
    )
    return {"source": "database", "data": user_data}

Advantages:
- Redis supports distributed deployment, allowing all service instances to share the same cache.
- setex sets an expiration time to prevent permanent memory usage by stale cache.
- Supports multiple data structures (strings, hashes, lists, etc.) for complex caching needs.

三、Summary: Global Variables vs. Caching¶

四、Practical Recommendations¶

• Development Environment: Use global variables (with locking) for simple scenarios and cachetools for high-frequency data.
• Production Environment: Use caching (e.g., Redis) for multi-process/distributed deployments to avoid cross-process issues with global variables.
• Data Security: Encrypt sensitive data in caches to prevent plaintext storage (e.g., passwords) in Redis.

With the examples provided, you now have the basic knowledge to manage state in FastAPI. Choose the appropriate solution based on your project needs to efficiently address data sharing and performance optimization.