Building High-Throughput Async Microservices with FastAPI and Redis

Building scalable, low-latency microservices requires non-blocking I/O and efficient memory management. Python’s asyncio ecosystem, combined with FastAPI and Redis, allows developers to handle thousands of concurrent requests per second on minimal infrastructure.

1. Asynchronous I/O Execution Model

Traditional synchronous frameworks (like Flask or Django WSGI) spawn dedicated threads per incoming request. When making database queries or external API calls, the thread blocks, idling CPU and memory resources.

FastAPI runs on an ASGI server (Uvicorn/Hypercorn) using a single-threaded event loop that handles context switching during I/O operations without blocking the main process:

graph TD;
    Client[Client Requests] --> EventLoop[ASGI Event Loop];
    EventLoop -->|Task A| RedisIO[Waiting for Redis I/O];
    EventLoop -->|Task B| CPUProcessing[Processing CPU Task];
    EventLoop -->|Task C| AsyncSQL[Executing Async SQL];

2. Implementing Redis Connection Pooling

Re-creating connection sockets per API endpoint call creates heavy overhead. Use an asynchronous connection pool managed via FastAPI’s lifespan events:

from fastapi import FastAPI, Depends from redis.asyncio import ConnectionPool, Redis from contextlib import asynccontextmanager # Global connection pool reference redis_pool: ConnectionPool = None @asynccontextmanager async def lifespan(app: FastAPI): global redis_pool # Initialize connection pool on startup redis_pool = ConnectionPool.from_url( "redis://localhost:6379/0", max_connections=20, decode_responses=True ) yield # Close connections on shutdown await redis_pool.disconnect() app = FastAPI(lifespan=lifespan) async def get_redis() -> Redis: return Redis(connection_pool=redis_pool)

3. High-Throughput Endpoint with Caching Pattern

Here is an example implementation of a read-through cache strategy with automatic key expiration:

from fastapi import HTTPException import json @app.get("/api/v1/analytics/{sensor_id}") async def get_sensor_analytics( sensor_id: str, db_redis: Redis = Depends(get_redis) ): cache_key = f"cache:sensor:{sensor_id}" # 1. Check Redis Cache cached_data = await db_redis.get(cache_key) if cached_data: return {"source": "cache", "data": json.loads(cached_data)} # 2. Cache Miss - Simulate DB Retrieval db_result = await fetch_from_database(sensor_id) if not db_result: raise HTTPException(status_code=404, detail="Sensor not found") # 3. Asynchronously Populate Cache (TTL = 300 seconds) await db_redis.setex( name=cache_key, time=300, value=json.dumps(db_result) ) return {"source": "database", "data": db_result} async def fetch_from_database(sensor_id: str): # Simulated DB call return {"sensor_id": sensor_id, "status": "active", "value": 42.8}

4. Key Takeaways

• Avoid Blocking Calls inside async def: Never use synchronous libraries (e.g., standard requests or psycopg2) directly inside async route handlers. Use httpx or asyncpg instead.
• Connection Limits: Set explicit max_connections on your Redis and database pools to avoid exhausting system file descriptors under high loads.