Build async connection pools with FastAPI lifespan

Introduction

When your GenAI service opens a fresh PostgreSQL connection on every RAG query, a new Redis client on every cache check, and a TLS-renegotiating HTTP client on every LLM call, a single Locust ramp-up will exhaust your database's max_connections and surface as cascading 500s across every endpoint. Teams that skip centralized pool management discover this the first time a rolling deployment drops in-flight batch writes mid-stream, leaving conversation history rows that never reached PostgreSQL even though the LLM response was already cached in Redis. By the end of this lesson you'll be able to implement an async ConnectionPoolManager that wires asyncpg, redis.asyncio, and httpx pools into FastAPI's lifespan hooks with health probes and reverse-dependency-ordered shutdown.

Key Terminology

• Connection pool: A bounded set of pre-established client connections (PostgreSQL, Redis, or HTTP) reused across requests to amortize handshake cost and cap concurrent backend load.
• Lifespan hook: FastAPI's asynccontextmanager-based startup/shutdown protocol that wraps the application lifecycle, guaranteeing teardown code runs even when startup partially fails.
• Reverse-dependency-ordered shutdown: A close sequence that drains pools in the opposite order of their data flow (HTTP → Redis → PostgreSQL), so in-flight responses can still write to downstream stores before those stores disappear.

Concepts

The two ideas below explain how to size pools for GenAI traffic patterns and how to drain them safely under Kubernetes-driven rolling updates.

Connection Pool Sizing for GenAI Workloads

Sizing pools incorrectly is the single most common cause of performance degradation in GenAI applications. The following guidelines tie pool sizes to the specific workload patterns covered throughout this chapter:

• PostgreSQL max_size: Set this to your expected concurrent request count divided by the average query duration ratio. If your Locust test simulates 100 concurrent users and each RAG query holds a connection for 50ms, you need at most 100 × 0.05 = 5 connections at steady state. The max_size=20 default provides 4x headroom for burst traffic during batch processing runs where multiple LLM responses write embeddings simultaneously.

• Redis max_connections: Prompt cache operations (both Anthropic explicit and OpenAI prefix caching lookups) are sub-millisecond, so each connection serves ~1000 operations per second. With 50 connections, you can sustain 50,000 cache checks per second—well above what even aggressive DSPy compilation workloads generate.

• HTTP max_connections: This is your primary throttle for LLM API parallelism. The batch request processor groups calls by provider and enforces concurrency limits through this pool. Setting max_connections=100 with max_keepalive_connections=20 means you can burst to 100 simultaneous LLM calls (useful during batch processing) while maintaining 20 warm connections for steady-state streaming requests.

• min_size for PostgreSQL: Pre-warming with min_size=5 eliminates the connection establishment penalty on the first 5 requests after startup. This matters particularly during Locust test ramp-up, where the first wave of simulated users would otherwise all block on connection creation simultaneously, skewing your P99 latency measurements.

Graceful Shutdown Under Load

When Kubernetes initiates a rolling update, your pod receives SIGTERM and has terminationGracePeriodSeconds (default 30s) to finish in-flight work. The ConnectionPoolManager.close() method handles this by closing pools in reverse dependency order, but you must also signal FastAPI to stop accepting new requests. The combination of Kubernetes preStop hook (a 5-second sleep that lets the Service endpoint update propagate) and the lifespan finally block ensures that in-flight batch LLM requests complete their writes to PostgreSQL before the pool closes. Without this ordering, you risk partial writes—an LLM response that was cached in Redis but whose conversation history entry never reached PostgreSQL, creating an inconsistency that surfaces as missing context in subsequent agent loop iterations.

The pattern described here integrates directly with the batch processing pipeline later in this chapter: the BatchRequestProcessor acquires connections from these managed pools, and the backpressure mechanism respects pool exhaustion by pausing queue consumption when pg_pool.get_size() >= pg_pool.get_max_size(). Similarly, the Locust test suites in the final section validate that the system degrades gracefully under connection pressure—reporting increased latency rather than hard failures when pools approach their limits.

Code Walkthrough

Now that you've seen how to size pools for GenAI traffic and drain them in reverse dependency order, the walkthrough below moves from motivation to implementation: first why a centralized manager is required at all, then the ConnectionPoolManager class itself, and finally how it plugs into FastAPI's lifespan hooks with a /health endpoint Kubernetes probes can hit.

Why Centralized Pool Management Matters for GenAI Stacks

The connection lifecycle in a GenAI application differs fundamentally from traditional web services. A single RAG query might acquire a PostgreSQL connection to fetch document embeddings, a Redis connection to check the prompt cache (as covered in the multi-provider caching section), and an HTTP connection to send the assembled prompt to Anthropic's API with cache_control headers. If any of these connections comes from an unmanaged source—created ad-hoc inside the request handler—you lose three critical properties:

• Bounded concurrency: Without pool size limits, a burst of Locust-simulated traffic can open hundreds of PostgreSQL connections, triggering too many connections errors that cascade into 500 responses across your entire API surface.
• Connection reuse: Each asyncpg.connect() call outside a pool performs DNS resolution, TCP handshake, and PostgreSQL authentication. Under the sustained load patterns you will design in the Locust testing section, this overhead compounds into seconds of cumulative latency per request batch.
• Graceful shutdown: When Kubernetes sends SIGTERM during a rolling deployment, in-flight LLM requests through your batch processor need their HTTP connections drained—not abruptly terminated. A centralized manager with shutdown hooks ensures every pool closes in the correct dependency order.

The following diagram illustrates how the ConnectionPoolManager sits between FastAPI's lifespan protocol and the three pool types, providing a single initialization and teardown path:

ConnectionPoolManager centralizes the lifecycle of three async connection backends—asyncpg.create_pool for PostgreSQL, redis.asyncio.ConnectionPool for caching, and httpx.AsyncClient for outbound HTTP—within FastAPI's lifespan hooks. By initializing all pools at startup and draining them in reverse dependency order at shutdown (HTTP → Redis → PostgreSQL), this architecture prevents connection leaks under high concurrency and ensures RAG endpoints, batch processors, and agent loops share warm connections instead of paying per-request handshake costs.

Loading diagram...

Notice the shutdown sequence flows in reverse dependency order: HTTP connections drain first (because they may still be writing results to Redis or PostgreSQL), then Redis closes, and finally PostgreSQL—the most stateful resource—shuts down last.

Building the ConnectionPoolManager

The core abstraction is a single class that owns all three pool instances and exposes async methods for lifecycle management. The following implementation defines ConnectionPoolManager with create_pg_pool(), create_redis_pool(), and create_http_client() factory methods, each accepting configuration parameters tuned for GenAI workloads. The asyncpg.create_pool() call uses min_size=5 to pre-warm connections at startup (avoiding cold-start latency on the first batch of DSPy compilation requests), max_size=20 to cap connections under Locust load, and command_timeout=30 to prevent long-running embedding queries from holding connections indefinitely. The setup callback on each new PostgreSQL connection installs the pgvector extension required for RAG similarity searches. The Redis pool uses max_connections=50 to handle the high-throughput prompt cache reads that the caching layer generates. The HTTP client configures limits with max_connections=100 and max_keepalive_connections=20 to balance LLM API parallelism against provider rate limits.

1import asyncpg
2import redis.asyncio as aioredis
3import httpx
4from dataclasses import dataclass, field
5from typing import Optional
6
7@dataclass
8class PoolConfig:
9    pg_dsn: str = "postgresql://user:pass@localhost:5432/genai"
10    pg_min_size: int = 5
11    pg_max_size: int = 20
12    pg_command_timeout: int = 30
13    redis_url: str = "redis://localhost:6379/0"
14    redis_max_connections: int = 50
15    http_max_connections: int = 100
16    http_max_keepalive: int = 20
17    http_timeout: float = 60.0
18
19class ConnectionPoolManager:
20    def __init__(self, config: PoolConfig):
21        self.config = config
22        self.pg_pool: Optional[asyncpg.Pool] = None
23        self.redis_pool: Optional[aioredis.ConnectionPool] = None
24        self.http_client: Optional[httpx.AsyncClient] = None
25        self._initialized: bool = False
26
27    async def _pg_setup_callback(self, connection: asyncpg.Connection):
28        await connection.execute("CREATE EXTENSION IF NOT EXISTS vector")
29        await connection.set_type_codec(
30            "jsonb", encoder=str, decoder=str, schema="pg_catalog"
31        )
32
33    async def create_pg_pool(self) -> asyncpg.Pool:
34        self.pg_pool = await asyncpg.create_pool(
35            dsn=self.config.pg_dsn,
36            min_size=self.config.pg_min_size,
37            max_size=self.config.pg_max_size,
38            command_timeout=self.config.pg_command_timeout,
39            setup=self._pg_setup_callback,
40        )
41        return self.pg_pool
42
43    async def create_redis_pool(self) -> aioredis.Redis:
44        self.redis_pool = aioredis.ConnectionPool.from_url(
45            self.config.redis_url,
46            max_connections=self.config.redis_max_connections,
47            decode_responses=True,
48        )
49        return aioredis.Redis(connection_pool=self.redis_pool)
50
51    async def create_http_client(self) -> httpx.AsyncClient:
52        self.http_client = httpx.AsyncClient(
53            limits=httpx.Limits(
54                max_connections=self.config.http_max_connections,
55                max_keepalive_connections=self.config.http_max_keepalive,
56            ),
57            timeout=httpx.Timeout(self.config.http_timeout),
58        )
59        return self.http_client
60
61    async def initialize(self):
62        await self.create_pg_pool()
63        await self.create_redis_pool()
64        await self.create_http_client()
65        self._initialized = True
66
67    async def close(self):
68        if self.http_client:
69            await self.http_client.aclose()
70        if self.redis_pool:
71            await self.redis_pool.aclose()
72        if self.pg_pool:
73            await self.pg_pool.close()
74        self._initialized = False

Integrating with FastAPI Lifespan Hooks

FastAPI's lifespan context manager replaced the deprecated on_event("startup") and on_event("shutdown") decorators starting in version 0.93. The lifespan approach is superior for pool management because it guarantees the teardown code runs even if startup partially fails—critical when your create_pg_pool() succeeds but create_redis_pool() raises a ConnectionRefusedError. The following implementation wires ConnectionPoolManager into FastAPI's lifespan, stores the manager on app.state for dependency injection into request handlers, and adds a /health endpoint that probes all three pools. This health endpoint is what Kubernetes liveness and readiness probes hit, and it is also used by Locust test suites to verify the system is ready before sending simulated GenAI traffic.

1from contextlib import asynccontextmanager
2from fastapi import FastAPI, Depends
3import logging
4
5logger = logging.getLogger(__name__)
6
7@asynccontextmanager
8async def lifespan(app: FastAPI):
9    config = PoolConfig()
10    manager = ConnectionPoolManager(config)
11    try:
12        await manager.initialize()
13        app.state.pool_manager = manager
14        app.state.pg_pool = manager.pg_pool
15        app.state.redis = await manager.create_redis_pool()
16        app.state.http_client = manager.http_client
17        logger.info("All connection pools initialized")
18        yield
19    finally:
20        logger.info("Shutting down connection pools")
21        await manager.close()
22        logger.info("All connection pools closed")
23
24app = FastAPI(lifespan=lifespan)
25
26@app.get("/health")
27async def health_check():
28    manager: ConnectionPoolManager = app.state.pool_manager
29    checks = {}
30    try:
31        async with manager.pg_pool.acquire() as conn:
32            await conn.fetchval("SELECT 1")
33        checks["postgresql"] = "ok"
34    except Exception as e:
35        checks["postgresql"] = f"error: {e}"
36    try:
37        pong = await app.state.redis.ping()
38        checks["redis"] = "ok" if pong else "error: no pong"
39    except Exception as e:
40        checks["redis"] = f"error: {e}"
41    try:
42        resp = await manager.http_client.get("https://api.anthropic.com/v1/models")
43        checks["http_client"] = "ok" if resp.status_code < 500 else "degraded"
44    except Exception as e:
45        checks["http_client"] = f"error: {e}"
46
47    all_ok = all(v == "ok" for v in checks.values())
48    status_code = 200 if all_ok else 503
49    return {"status": "healthy" if all_ok else "degraded", "checks": checks}

Do's and Don'ts

Do's

1. ✓Do initialize all three pools inside FastAPI's lifespan startup hook via ConnectionPoolManager.initialize() — pools created ad-hoc inside request handlers lack bounded concurrency, so a Locust ramp-up can exhaust PostgreSQL's max_connections and cascade into 500s across every RAG and agent-loop endpoint simultaneously.
2. ✓Do set min_size=5 on asyncpg.create_pool() to pre-warm PostgreSQL connections at startup — without it, the first burst of RAG embedding queries pays DNS resolution, TCP handshake, and PostgreSQL authentication costs per connection, compounding into seconds of cumulative latency before the pool has time to fill.
3. ✓Do drain pools in reverse dependency order during shutdown (HTTP → Redis → PostgreSQL) — closing httpx.AsyncClient first ensures in-flight LLM responses finish writing before Redis and PostgreSQL close; inverting the order leaves conversation history rows mid-write when Kubernetes sends SIGTERM during a rolling deployment.

Don'ts

1. ✗Don't omit command_timeout=30 from asyncpg.create_pool() — without it, long-running embedding queries hold connections indefinitely, starving concurrent RAG endpoints of pool slots until the pool deadlocks under sustained Locust load while max_size=20 is already saturated.
2. ✗Don't share a single httpx.AsyncClient instance configured without limits — without max_connections=100 and max_keepalive_connections=20, the client either opens unbounded TLS connections to the LLM provider (hitting rate limits) or serializes all outbound requests through a single keepalive socket, bottlenecking your batch processor throughput.
3. ✗Don't create a redis.asyncio.ConnectionPool per request or per module import — instantiating pools outside ConnectionPoolManager means each import path holds its own set of up to max_connections=50 sockets, multiplying Redis connections beyond what the cache node can handle and bypassing the centralized shutdown hook that guarantees clean drain during lifespan teardown.