redis.asyncio and Connection Pooling

Overview

This is module 4's most important technical capsule from a production perspective. You'll learn to use redis.asyncio correctly — not just "putting an await on everything," but configuring the connection pool with the right parameters, handling reconnection, and sharing the pool across your whole application with the singleton pattern. Without proper pooling, an API with 100 concurrent users can exhaust Redis's connections and go down completely. With proper pooling, you handle thousands of concurrent requests with 50 reused physical connections.

Conceptually it's simple: a connection pool is a pre-created group of connections your application reuses. Instead of opening a new connection for every operation (with its round-trip of TCP handshake + auth), you take one from the pool, use it, and give it back. Under the hood, redis-py does this automatically when you create a client — but the defaults are conservative, and you need to tune them for your use case. An app with 4 workers and 1000 default connections = 4000 connections against Redis (which has a default limit of 10,000).

You'll cover how to configure max_connections appropriately, what to do when the pool is exhausted (ConnectionError: max number of clients reached), how to implement automatic reconnection for resilience, and the singleton pattern so your whole application shares the same pool. By the end you'll have the technical scaffolding that capsule 04 (FastAPI integration) and all of module 5 take for granted.

redis.asyncio: the modern client

The correct import

# ✅ Correct: the modern client (redis-py 4.2+, 5.x)
from redis.asyncio import Redis, ConnectionPool

# ❌ Incorrect: a library deprecated since 2021
import aioredis

redis.asyncio is a module inside redis-py. It isn't a separate library. If your pip show redis shows version >= 4.2, you already have redis.asyncio available.

An API identical to the sync one — you just add await

import asyncio
from redis.asyncio import Redis


async def main():
    r = Redis(host='localhost', port=6379, decode_responses=True)

    # Every operation is the same — just with await
    await r.set("key", "value")
    value = await r.get("key")
    print(value)

    # Hashes
    await r.hset("user:1", mapping={"name": "Alice", "age": "30"})
    profile = await r.hgetall("user:1")
    print(profile)

    # Sorted sets
    await r.zadd("leaderboard", {"alice": 1500, "bob": 2300})
    top = await r.zrevrange("leaderboard", 0, 4, withscores=True)
    print(top)

    # Pub/Sub (you saw it in capsule 02)
    await r.publish("news", "hello")

    # Cleanup
    await r.aclose()


asyncio.run(main())

There's no new API. Every command from module 1 works identically, just with await.

aclose() vs close()

• await r.aclose() — closes the async connection (preferred for redis.asyncio)
• r.close() (sync) — available but deprecated in redis.asyncio

Use aclose() to avoid warnings.

Async pipelines

Pipelines also work with await:

async with r.pipeline() as pipe:
    pipe.set("key1", "v1")
    pipe.set("key2", "v2")
    pipe.get("key1")
    results = await pipe.execute()

print(results)  # [True, True, "v1"]

Useful for batch operations without multiple round-trips. Just like the sync version from module 1's capsule 5.

The Connection Pool: why it's critical

The problem without pooling

When you create Redis(...) directly, a pool is created internally — but the defaults are:

max_connections: unlimited (it will create as many as you ask for)

Imagine 1,000 concurrent requests each doing await r.get():

• Without sharing a pool: each Redis(...) creates its own pool. 1000 instances = 1000 pools = potentially 1000+ connections
• With a single unlimited pool: 1000 simultaneous connections to Redis

Redis has a default limit of 10,000 max clients (maxclients in redis.conf). An app with 4 uvicorn workers and 1000 reqs/sec each can saturate Redis fast.

Solving it with max_connections

from redis.asyncio import Redis, ConnectionPool


# A pool with an explicit limit
pool = ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
    max_connections=50,
)

r = Redis(connection_pool=pool)

Now a maximum of 50 physical connections are kept in the pool. When 100 requests arrive at the same time:

• The first 50 take connections from the pool
• The next 50 wait briefly until a connection is released
• Total: 50 physical connections, not 100

How many connections you need

The calculation rule:

max_connections = (average_concurrent_requests / requests_per_connection_per_second) × safety_factor

With real numbers:

• The API handles 1000 req/sec
• Each request does 2-3 Redis ops (get + set)
• Each op takes ~1 ms
• One connection can do ~500 ops/sec
• requests_per_connection = 500 / 3 = 166

max_connections = (1000 / 166) × 1.5 (safety) = ~10

For most typical apps, 50-100 max_connections is enough. If you see "pool exhausted" errors, raise it. But max_connections=10000 is ridiculous (you're probably caching badly or you have a leak).

Symptoms of a badly configured pool

A pool that's too small:

redis.exceptions.ConnectionError: Too many connections

It means: a lot of concurrent requests waiting for a connection. The solution: raise max_connections or reduce usage (caching).

A pool that's too large:

redis.exceptions.ConnectionError: max number of clients reached

It means: your app OPENED more connections than Redis accepts. The solution: lower max_connections or raise maxclients in redis.conf.

The singleton pattern for sharing the pool

In a FastAPI application with multiple files, the pool MUST be a singleton. If every module creates its own Redis(...), the pools pile up.

Implementing the singleton

Create redis_client.py:

"""
A singleton for the async Redis client.
"""
from redis.asyncio import Redis, ConnectionPool


_pool: ConnectionPool | None = None
_client: Redis | None = None


def init_pool(
    host: str = "localhost",
    port: int = 6379,
    db: int = 0,
    max_connections: int = 50,
    socket_timeout: float = 2.0,
    socket_connect_timeout: float = 2.0,
):
    """Initializes the global pool (call it once at startup)."""
    global _pool, _client

    _pool = ConnectionPool(
        host=host,
        port=port,
        db=db,
        decode_responses=True,
        max_connections=max_connections,
        socket_timeout=socket_timeout,
        socket_connect_timeout=socket_connect_timeout,
    )

    _client = Redis(connection_pool=_pool)


def get_redis() -> Redis:
    """Returns the global instance of the Redis client."""
    if _client is None:
        raise RuntimeError("Redis pool not initialized. Call init_pool() first.")
    return _client


async def close_pool():
    """Closes the pool when the app shuts down."""
    global _pool, _client
    if _client:
        await _client.aclose()
        _client = None
    if _pool:
        await _pool.aclose()
        _pool = None

Using it from other modules

# auth.py
from redis_client import get_redis


async def create_session(user_id: int):
    r = get_redis()  # the same pool, the same client
    await r.hset(f"session:{...}", ...)


# rate_limiter.py
from redis_client import get_redis


async def check_rate_limit(user_id: str):
    r = get_redis()  # the SAME pool — no extra overhead
    # ...


# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from redis_client import init_pool, close_pool


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_pool(max_connections=50)
    yield
    await close_pool()


app = FastAPI(lifespan=lifespan)

The singleton's advantages

1. A single connection pool for the whole app — efficient
2. A clean shutdown — no connection leaks
3. Easy to test — you mock get_redis() in tests
4. Centralized configuration — a single place to change the parameters

A connection pool with a URL

In production, you configure Redis with a URL (from environment variables). redis-py accepts URLs:

from redis.asyncio import ConnectionPool, Redis
import os


REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")

# Create the pool from a URL
pool = ConnectionPool.from_url(
    REDIS_URL,
    max_connections=50,
    decode_responses=True,
)

r = Redis(connection_pool=pool)

URL formats

redis://localhost:6379/0                          # local, no auth
redis://default:password@redis-host:6379/0        # with a password
rediss://default:password@redis-host:6379/0       # with TLS (SSL)
redis://localhost:6379/0?max_connections=100      # query params

In the cloud, typically:

# AWS ElastiCache
REDIS_URL="redis://my-cluster.abc123.use1.cache.amazonaws.com:6379"

# Upstash
REDIS_URL="rediss://default:abc123@my-redis.upstash.io:6379"

# Redis Cloud
REDIS_URL="redis://default:abc123@redis-12345.cloud.redislabs.com:12345"

from_url() parses the format automatically.

Health checks

In production, verify that Redis is available. This catches problems before they affect users.

async def check_redis_health() -> dict:
    """A health check for Redis. It returns a detailed status."""
    r = get_redis()

    try:
        # Ping with a timeout
        result = await asyncio.wait_for(r.ping(), timeout=1.0)
        if result:
            # Check additional info
            info = await r.info("server")
            return {
                "status": "healthy",
                "version": info.get("redis_version"),
                "uptime_seconds": info.get("uptime_in_seconds"),
            }
        return {"status": "unhealthy", "reason": "ping returned False"}
    except asyncio.TimeoutError:
        return {"status": "unhealthy", "reason": "ping timeout"}
    except Exception as e:
        return {"status": "unhealthy", "reason": str(e)}

A health check endpoint in FastAPI

@app.get("/health")
async def health():
    redis_health = await check_redis_health()

    overall = "healthy" if redis_health["status"] == "healthy" else "degraded"

    return {
        "status": overall,
        "services": {
            "redis": redis_health,
        }
    }

Health checks in production

Your orchestrator (Kubernetes, ECS, etc.) polls /health periodically. If it returns a 503, the orchestrator can:

• Restart the container
• Pull the container out of the load balancer
• Alert the team

@app.get("/health")
async def health():
    redis_health = await check_redis_health()

    if redis_health["status"] != "healthy":
        # Return a 503 so the orchestrator detects it
        raise HTTPException(503, "Redis unavailable")

    return {"status": "healthy"}

Automatic reconnection

If Redis is temporarily down, your pool will try to reconnect automatically. But there are settings that make this behavior more resilient.

Retry on timeout

from redis.backoff import ExponentialBackoff
from redis.asyncio.retry import Retry


pool = ConnectionPool.from_url(
    REDIS_URL,
    max_connections=50,
    retry=Retry(ExponentialBackoff(cap=10, base=1), 3),  # 3 retries with backoff
    retry_on_timeout=True,
)

If an operation fails with a timeout:

• It waits 1 second, retries
• It waits 2 seconds, retries
• It waits 4 seconds, retries
• If all 3 fail, it raises

The pool's health check

pool = ConnectionPool.from_url(
    REDIS_URL,
    max_connections=50,
    health_check_interval=30,  # ping every 30s to keep the connections alive
)

health_check_interval=30 makes every idle connection in the pool get "pinged" every 30 seconds. It detects and discards dead connections (e.g., NAT timeouts in the cloud).

Graceful degradation

When Redis fails, your app should degrade gracefully — not die.

import logging
from redis.exceptions import RedisError


logger = logging.getLogger(__name__)


async def get_cached_or_fallback(key: str, fallback_fn):
    """
    Tries the cache; if Redis fails, it goes to the fallback (the DB) silently.
    """
    r = get_redis()

    try:
        cached = await r.get(key)
        if cached:
            return json.loads(cached)
    except RedisError as e:
        logger.warning(f"Redis read failed for {key}: {e}")
        # No raise — degrade to the fallback

    # A cache miss or Redis is down: go to the source
    data = await fallback_fn()

    try:
        await r.set(key, json.dumps(data), ex=300)
    except RedisError as e:
        logger.warning(f"Redis write failed for {key}: {e}")
        # No raise — the data is returned anyway

    return data

The pattern in endpoints

@app.get("/products/{id}")
async def get_product(id: int):
    return await get_cached_or_fallback(
        key=f"product:{id}",
        fallback_fn=lambda: db.get_product(id),
    )

If Redis goes down:

• ✗ The latency climbs (every request goes to the DB)
• ✓ The API KEEPS WORKING

The logs show warnings, but the user doesn't see errors.

Testing the pool under concurrency

Let's verify the pool works under real load.

"""
Test: 200 concurrent requests with max_connections=10.
If the pool works, they all finish successfully.
"""
import asyncio
import time
from redis.asyncio import Redis, ConnectionPool


async def make_request(r, i):
    try:
        await r.set(f"test:{i}", f"value-{i}")
        v = await r.get(f"test:{i}")
        return v == f"value-{i}"
    except Exception as e:
        return f"error: {e}"


async def main():
    pool = ConnectionPool(
        host='localhost',
        port=6379,
        decode_responses=True,
        max_connections=10,  # only 10!
    )
    r = Redis(connection_pool=pool)

    # 200 concurrent requests with only 10 connections
    start = time.time()
    tasks = [make_request(r, i) for i in range(200)]
    results = await asyncio.gather(*tasks)
    elapsed = time.time() - start

    success = sum(1 for r in results if r is True)
    errors = [r for r in results if r is not True]

    print(f"Total: 200 requests")
    print(f"Success: {success}")
    print(f"Errors: {len(errors)}")
    print(f"Time: {elapsed:.2f}s")

    if errors:
        print(f"First error: {errors[0]}")

    await r.aclose()
    await pool.aclose()


asyncio.run(main())

Expected output:

Total: 200 requests
Success: 200
Errors: 0
Time: 0.45s

The 200 requests ran successfully with only 10 physical connections. The pool handles the concurrency transparently.

Decode responses: when to and when not to

# decode_responses=True (the recommended default)
r = Redis(decode_responses=True)
v = await r.get("key")  # str

# decode_responses=False (rare, for special cases)
r = Redis(decode_responses=False)
v = await r.get("key")  # bytes

Use decode_responses=True for:

• Normal strings
• Serialized JSON
• Numbers
• Any text

Use decode_responses=False for:

• Binary data (files, images)
• Pickle bytes
• Compression (gzip, etc.)

For 99% of cases, decode_responses=True is what you want.

Troubleshooting

Problem 1: RuntimeError: This event loop is already running

Cause: You're calling asyncio.run() inside a context that already has an event loop (Jupyter, a misconfigured FastAPI test client).

Solution:

# In Jupyter:
import nest_asyncio
nest_asyncio.apply()

await main()  # await directly, not asyncio.run()

Problem 2: ConnectionError: Too many connections

Cause: Your app exceeds max_connections or Redis's maxclients.

Solution:

1. Raise the pool's max_connections:

   pool = ConnectionPool(..., max_connections=100)

2. Check Redis's maxclients:

   redis-cli CONFIG GET maxclients
   # 1) "maxclients"
   # 2) "10000"

3. Check the current connections:

   redis-cli INFO clients | grep connected_clients
   # connected_clients:5

4. If you have a leak (more connections than expected), audit it:

   redis-cli CLIENT LIST

Problem 3: The pool works locally but fails in production

Cause: Network latency. On localhost: 0.1 ms. In the cloud: 1-50 ms.

Solution:

1. Raise the timeouts:

   pool = ConnectionPool(
       socket_timeout=5.0,           # more tolerant (the default is 2)
       socket_connect_timeout=5.0,
   )

2. A health check interval:

   pool = ConnectionPool(health_check_interval=30)

3. Consider a Redis local to the server or a regional cluster to cut latency

Problem 4: An await r.aclose() warning

Cause: You're using r.close() (sync) instead of await r.aclose() (async).

Solution:

# ❌ Wrong
r.close()

# ✅ Right
await r.aclose()

Problem 5: The pool works in one test but not another

Cause: The singleton isn't initialized between tests.

Solution: Initialize the pool in a shared fixture:

import pytest


@pytest.fixture(scope="session")
async def redis_client():
    init_pool()
    yield get_redis()
    await close_pool()


@pytest.mark.asyncio
async def test_something(redis_client):
    await redis_client.set("test", "value")
    # ...

Problem 6: Connections that close after minutes without use

Cause: A NAT timeout (in the cloud), firewalls, or an idle connection timeout in Redis.

Solution:

pool = ConnectionPool(
    socket_keepalive=True,
    health_check_interval=30,  # ping idle connections
)

Exercises

Exercise 1: A basic async client (Easy)

Write a script that connects with redis.asyncio, does set/get/hset/hgetall/zadd/zrange, and shuts down cleanly with aclose().

import asyncio
from redis.asyncio import Redis


async def main():
    r = Redis(host='localhost', port=6379, decode_responses=True)

    # Verify the connection
    pong = await r.ping()
    print(f"Ping: {pong}")

    # String ops
    await r.set("greeting", "hello")
    print(f"Get: {await r.get('greeting')}")

    # Hash ops
    await r.hset("user:1", mapping={"name": "Alice", "age": "30"})
    print(f"Hash: {await r.hgetall('user:1')}")

    # Sorted set ops
    await r.zadd("scores", {"alice": 100, "bob": 200, "charlie": 150})
    top = await r.zrevrange("scores", 0, -1, withscores=True)
    print(f"Top: {top}")

    # Cleanup
    await r.aclose()


asyncio.run(main())

Ping: True
Get: hello
Hash: {'name': 'Alice', 'age': '30'}
Top: [('bob', 200.0), ('charlie', 150.0), ('alice', 100.0)]

Exercise 2: A pool with max_connections=5 (Medium)

Create a pool with max_connections=5 and fire 50 concurrent requests. Verify all 50 finish successfully (the pool handles the queueing).

import asyncio
import time
from redis.asyncio import Redis, ConnectionPool


async def operation(r, i):
    await r.set(f"key:{i}", f"value-{i}")
    val = await r.get(f"key:{i}")
    return val == f"value-{i}"


async def main():
    pool = ConnectionPool(
        host='localhost',
        port=6379,
        decode_responses=True,
        max_connections=5,
    )
    r = Redis(connection_pool=pool)

    start = time.time()
    tasks = [operation(r, i) for i in range(50)]
    results = await asyncio.gather(*tasks)
    elapsed = time.time() - start

    print(f"Successful: {sum(results)}/50")
    print(f"Time: {elapsed:.2f}s with max_connections=5")

    await r.aclose()
    await pool.aclose()


asyncio.run(main())

Successful: 50/50
Time: 0.15s with max_connections=5

Exercise 3: The singleton pattern (Medium)

Implement the redis_client.py singleton (init_pool, get_redis, close_pool). Create 3 files (auth.py, rate.py, cache.py) that use get_redis(). Verify they all share the same client.

# redis_client.py
from redis.asyncio import Redis, ConnectionPool


_pool: ConnectionPool | None = None
_client: Redis | None = None


def init_pool(host="localhost", port=6379, max_connections=50):
    global _pool, _client
    _pool = ConnectionPool(
        host=host, port=port, decode_responses=True,
        max_connections=max_connections,
    )
    _client = Redis(connection_pool=_pool)


def get_redis() -> Redis:
    if _client is None:
        raise RuntimeError("Pool not initialized")
    return _client


async def close_pool():
    global _pool, _client
    if _client:
        await _client.aclose()
    if _pool:
        await _pool.aclose()
    _pool = None
    _client = None

# auth.py
from redis_client import get_redis


async def auth_op():
    r = get_redis()
    await r.set("auth:test", "ok")
    print(f"Auth uses client id: {id(r)}")


# rate.py
from redis_client import get_redis


async def rate_op():
    r = get_redis()
    await r.set("rate:test", "ok")
    print(f"Rate uses client id: {id(r)}")


# cache.py
from redis_client import get_redis


async def cache_op():
    r = get_redis()
    await r.set("cache:test", "ok")
    print(f"Cache uses client id: {id(r)}")

# main.py
import asyncio
from redis_client import init_pool, close_pool, get_redis
from auth import auth_op
from rate import rate_op
from cache import cache_op


async def main():
    init_pool(max_connections=20)

    # All use same client
    await auth_op()
    await rate_op()
    await cache_op()

    # Verify it's the same one
    r1 = get_redis()
    r2 = get_redis()
    print(f"\nSame client: {r1 is r2}")
    print(f"Pool max_connections: {r1.connection_pool.max_connections}")

    await close_pool()


asyncio.run(main())

Auth uses client id: 4378298992
Rate uses client id: 4378298992
Cache uses client id: 4378298992

Same client: True
Pool max_connections: 20

Exercise 4: A health check endpoint (Medium)

Implement check_redis_health() that returns the status with version and uptime info. Integrate it into a FastAPI /health endpoint. Test it with Redis up and down.

# health.py
import asyncio
from redis.asyncio import Redis
from redis.exceptions import RedisError


async def check_redis_health(r: Redis) -> dict:
    try:
        result = await asyncio.wait_for(r.ping(), timeout=1.0)
        if not result:
            return {"status": "unhealthy", "reason": "ping returned False"}

        info = await r.info("server")
        return {
            "status": "healthy",
            "version": info.get("redis_version"),
            "uptime_seconds": info.get("uptime_in_seconds"),
            "connected_clients": (await r.info("clients")).get("connected_clients"),
        }
    except asyncio.TimeoutError:
        return {"status": "unhealthy", "reason": "ping timeout"}
    except RedisError as e:
        return {"status": "unhealthy", "reason": f"redis error: {e}"}
    except Exception as e:
        return {"status": "unhealthy", "reason": f"unknown: {e}"}

# app.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from redis_client import init_pool, close_pool, get_redis
from health import check_redis_health


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_pool()
    yield
    await close_pool()


app = FastAPI(lifespan=lifespan)


@app.get("/health")
async def health():
    redis_status = await check_redis_health(get_redis())

    if redis_status["status"] != "healthy":
        raise HTTPException(503, detail={"redis": redis_status})

    return {
        "status": "healthy",
        "services": {"redis": redis_status},
    }

uvicorn app:app --reload &
sleep 2

curl -s http://localhost:8000/health | jq
# {
#   "status": "healthy",
#   "services": {
#     "redis": {
#       "status": "healthy",
#       "version": "7.2.4",
#       "uptime_seconds": 12345,
#       "connected_clients": 1
#     }
#   }
# }

# Stop Redis
docker stop redis-dev

curl -s http://localhost:8000/health
# 503 Service Unavailable
# {"detail":{"redis":{"status":"unhealthy","reason":"ping timeout"}}}

docker start redis-dev

Exercise 5: Graceful degradation (Medium-Hard)

Implement get_cached_or_fallback(key, fallback_fn) that uses Redis when it's up and falls back when Redis fails. Test: stop Redis, and verify the endpoint keeps working (more slowly).

import json
import logging
from redis.exceptions import RedisError
from redis_client import get_redis


logger = logging.getLogger(__name__)


async def get_cached_or_fallback(key: str, fallback_fn, ttl: int = 300):
    r = get_redis()

    # Try cache
    try:
        cached = await r.get(key)
        if cached:
            return json.loads(cached)
    except RedisError as e:
        logger.warning(f"Redis read failed for {key}: {e}")
        # Continue to fallback

    # Cache miss or Redis down: fallback
    data = await fallback_fn()

    # Try caching (best-effort)
    try:
        await r.set(key, json.dumps(data), ex=ttl)
    except RedisError as e:
        logger.warning(f"Redis write failed for {key}: {e}")
        # No raise

    return data

# Test
import asyncio
from redis_client import init_pool, close_pool


# We simulate the DB
async def db_get_product(pid):
    print(f"  [DB query for product:{pid}]")
    await asyncio.sleep(0.05)  # simulate DB latency
    return {"id": pid, "name": f"Product {pid}", "price": 100 + pid}


async def main():
    init_pool()

    # 1. With Redis up
    print("=== Redis UP ===")
    p1 = await get_cached_or_fallback(
        "product:1",
        lambda: db_get_product(1),
    )
    print(f"  Result: {p1}")

    # A second call: a cache hit
    p1_again = await get_cached_or_fallback(
        "product:1",
        lambda: db_get_product(1),
    )
    print(f"  Result (cached): {p1_again}")

    # 2. Stop Redis by hand:
    # docker stop redis-dev

    print("\n=== After stopping Redis (do it now) ===")
    print("Press Enter when Redis is down...")
    input()

    p2 = await get_cached_or_fallback(
        "product:2",
        lambda: db_get_product(2),
    )
    print(f"  Result (Redis down): {p2}")

    # Notice: the call SUCCEEDED even with Redis down

    await close_pool()


asyncio.run(main())

=== Redis UP ===
  [DB query for product:1]
  Result: {'id': 1, 'name': 'Product 1', 'price': 101}
  Result (cached): {'id': 1, 'name': 'Product 1', 'price': 101}

=== After stopping Redis ===
Press Enter when Redis is down...
[Enter]
WARNING: Redis read failed for product:2: Connection refused
  [DB query for product:2]
WARNING: Redis write failed for product:2: Connection refused
  Result (Redis down): {'id': 2, 'name': 'Product 2', 'price': 102}

Exercise 6: Pool exhaustion (Hard)

Create a pool with max_connections=2. Fire 100 concurrent requests, each doing 100 ms of operations. Measure the total time. Compare with max_connections=50.

import asyncio
import time
from redis.asyncio import Redis, ConnectionPool


async def slow_op(r, i):
    """A slow operation: a pipeline + a simulated sleep."""
    await r.set(f"slow:{i}", f"data-{i}")
    await asyncio.sleep(0.1)  # 100ms work
    val = await r.get(f"slow:{i}")
    return val == f"data-{i}"


async def benchmark(max_connections):
    pool = ConnectionPool(
        host='localhost',
        port=6379,
        decode_responses=True,
        max_connections=max_connections,
    )
    r = Redis(connection_pool=pool)

    start = time.time()
    tasks = [slow_op(r, i) for i in range(100)]
    await asyncio.gather(*tasks)
    elapsed = time.time() - start

    await r.aclose()
    await pool.aclose()

    return elapsed


async def main():
    # With a very small pool
    print("Benchmark with max_connections=2...")
    t1 = await benchmark(2)
    print(f"  Time: {t1:.2f}s")

    # With a reasonable pool
    print("\nBenchmark with max_connections=50...")
    t2 = await benchmark(50)
    print(f"  Time: {t2:.2f}s")

    print(f"\nSpeedup: {t1/t2:.1f}x with the bigger pool")


asyncio.run(main())

Benchmark with max_connections=2...
  Time: 5.50s

Benchmark with max_connections=50...
  Time: 0.45s

Speedup: 12.2x with the bigger pool

Summary

In this capsule you learned:

redis.asyncio:

• The modern client (redis-py 4.2+)
• An API identical to the sync one, with await
• aclose() instead of close()
• Pipelines and every operation work asynchronously

The Connection Pool:

• It reuses connections efficiently
• max_connections tunable to the expected concurrency
• Shared across the whole app with the singleton pattern
• A common error: a pool that's too small = "Too many connections"

The singleton pattern:

• init_pool() on startup, close_pool() on shutdown
• get_redis() returns the shared client
• Centralized configuration
• Easy to test

Advanced configuration:

• from_url() for configuration from environment variables
• health_check_interval to ping idle connections
• socket_keepalive for long-lived connections
• Retry with ExponentialBackoff for resilience

Health checks:

• r.ping() to verify connectivity
• A /health endpoint that reports a detailed status
• A 503 status code if Redis is down (so the orchestrator detects it)

Graceful degradation:

• Try/except around Redis operations
• A fallback to the DB when Redis fails
• Logging warnings without crashing the API
• The app works (more slowly) without Redis

The 2026 stack: redis>=7.4, redis.asyncio.Redis, redis.asyncio.ConnectionPool — NOT the deprecated aioredis.

Additional resources

1. redis.asyncio API Reference — Official docs with examples
2. Connection Pooling in Production — Best practices from Redis Labs
3. redis-py Migration from aioredis — The official migration guide
4. FastAPI Lifespan Events — The pattern for initializing resources
5. The Twelve-Factor App: Config — Config via environment variables
6. Redis Best Practices: Connection Management — The official recommendations

What's next?

In Capsule 04 you integrate everything you've learned into FastAPI professionally: dependency injection with Depends(get_redis), lifespan events, automatic caching middleware that intercepts and caches responses, and full graceful degradation. It's the integration pattern module 5 (the capstone project) assumes from day 1.

Keep Redis running. Let's go.