Module 4: Pub/Sub and FastAPI Integration

Introduction to Module 4: Pub/Sub and FastAPI Integration

Overview

You've closed 3 complete modules. You know how to cache with judgment, protect your API with professional rate limiting, and handle sessions with immediate revocation. But so far, everything you've written is synchronous. Every Redis operation blocks the thread until it finishes. Your API could handle more concurrency if the Redis operations were async — and FastAPI was designed precisely for that.

This module closes that gap. You'll switch from redis.Redis (sync) to redis.asyncio.Redis (async) — the official client from redis-py 4.2+. NOT the separate aioredis library, which has been deprecated since 2021. If you see tutorials importing aioredis, they're out of date — the modern client is redis.asyncio, which is included when you run pip install redis.

Then you'll learn Pub/Sub, Redis's lightweight messaging system. With a PUBLISH you send an event; with a SUBSCRIBE you listen for it somewhere else. It's perfect for cache invalidation events between components — when a value changes, you publish an event and every subscriber invalidates its related caches. But Pub/Sub has limitations you'll come to understand clearly: if nobody's listening when you publish, the message is lost. There's no persistence and no replay. For real durable messaging you need RabbitMQ or Kafka — but for the typical cases of invalidation and lightweight notifications, Redis Pub/Sub is the right tool.

The last 30% of the module is the professional integration with FastAPI: dependency injection with Depends(get_redis), automatic caching middleware, connection pooling, lifespan events to initialize the pool, health checks, and graceful degradation when Redis is down. It's the pattern you'll use in module 5's capstone project — the last piece before assembling the Production Cached API.

Where are we in the guide?

You're in Module 4 of 5 of the Redis & Caching Strategies guide:

Module 1: Redis Fundamentals ✅
    → Installation, CLI, the 5 data types, redis-py (sync)

Module 2: Caching Patterns & TTL ✅
    → Cache-aside, write-through, write-behind, TTL, invalidation

Module 3: Rate Limiting & Sessions ✅
    → Token/leaky bucket, sliding window, JWT + Redis sessions

Module 4: Pub/Sub & FastAPI Integration ← YOU ARE HERE
    → Async Redis, Pub/Sub, dependency injection, middleware, pooling

Module 5: Project — Production Cached API
    → The full stack: caching + rate limiting + sessions + Pub/Sub + metrics

How this module builds on the previous ones

From module 1: synchronous redis-py — you'll learn it's exactly the same, but with an await on every operation. If you command r.set(), you'll command await r.set() after 5 minutes.

From module 2: event-driven cache invalidation — a preview of Pub/Sub. Here you implement it for real. When a product changes, you publish cache:invalidate:product:42; the subscribers delete the related cache keys on their servers.

From module 3: you already saw FastAPI middleware with rate limiting. Here you extend the pattern with automatic caching middleware that intercepts responses and caches/serves them.

The connection with earlier guides in the path:

FastAPI dependency injection (Guide #7) — Depends(get_redis) is exactly the pattern you learned
Basic async/await in Python

Problem 1: sync vs async in FastAPI

FastAPI is an async-first framework. When you declare an async def endpoint, the framework can handle thousands of concurrent requests in a single worker because while one request waits on I/O (a DB query, a Redis op, an external HTTP call), the others can make progress.

But there's an anti-pattern that kills this benefit: synchronous operations inside async endpoints.

An example of the problem

# ❌ Wrong: it blocks the event loop
import redis
r = redis.Redis()  # SYNC

@app.get("/products/{id}")
async def get_product(id: int):
    cached = r.get(f"product:{id}")  # ← this blocks for ~1ms
    if cached:
        return json.loads(cached)
    # ...

The synchronous r.get() stops the event loop for the duration of the round-trip to Redis (~1 ms). During that 1 ms, no other request can make progress, even if they're ready.

With 1,000 concurrent requests hitting the same endpoint:

Sync: 1 ms × 1000 = 1 second total (the requests serialize)
Async: ~5 ms total (all 1,000 run concurrently)

Using sync in an async endpoint is 200x worse.

The solution: `redis.asyncio`

# ✅ Right: it doesn't block
import redis.asyncio as redis_async

r = redis_async.Redis()

@app.get("/products/{id}")
async def get_product(id: int):
    cached = await r.get(f"product:{id}")  # ← await, no blocking
    if cached:
        return json.loads(cached)
    # ...

You just add await and change the import. The client's syntax is identical to the sync one. There's no new API to learn.

Problem 2: communication between components

So far, all your code lives in a single FastAPI app. But in real production, systems have multiple components:

An API service (the one you've built)
An async worker that processes jobs (background tasks)
Cron jobs that clean up old sessions or caches
Other microservices in the same stack

How do these components talk to each other when something changes? For example:

An admin updates a product from an admin panel (another service)
How does the API service find out so it can invalidate its cache?
How does the worker that's caching reports find out?

Without Pub/Sub: the services get coupled. The API would have to call "InvalidateCache" on the worker, on the other services, etc. N services = N×N connections = chaos.

With Pub/Sub: the admin publishes product:42:changed. Every subscribed service receives the event simultaneously and reacts. Total decoupling.

┌─────────────┐
│ Admin       │
│ Service     │
└──────┬──────┘
       │ PUBLISH product:42:changed
       ▼
┌──────────────┐
│   Redis      │
│   Pub/Sub    │
└──┬───┬───┬───┘
   │   │   │
   ▼   ▼   ▼
┌─────┐ ┌──────┐ ┌────────┐
│ API │ │Worker│ │ Cron   │
│     │ │      │ │ Cleanup│
└─────┘ └──────┘ └────────┘

Each one reacts without knowing about the others.

What you'll learn

By the end of the 5 capsules:

Pub/Sub (capsule 02)

The basic commands: PUBLISH, SUBSCRIBE, PSUBSCRIBE (pattern matching), UNSUBSCRIBE
Use cases: cache invalidation events, real-time notifications, event broadcasting
Critical limitations: fire-and-forget, no persistence, no acknowledgment, no replay
When Pub/Sub is enough vs when you need RabbitMQ/Kafka

`redis.asyncio` and Connection Pooling (capsule 03)

The official async client: from redis.asyncio import Redis (NOT from aioredis import Redis)
Operations equivalent to the sync ones, but with await
The connection pool: ConnectionPool with max_connections, timeout, health checks
Why pooling is critical under concurrency (don't open 1000 connections for 1000 requests)
The singleton pattern for sharing the pool across every worker

Deep FastAPI Integration (capsule 04)

Dependency injection: async def get_redis() with Depends(get_redis)
Lifespan events: initializing the pool on startup, cleaning up on shutdown
Automatic caching middleware: it intercepts responses, caches them, and respects Cache-Control headers
A health check endpoint: verifying connectivity to Redis
Graceful degradation: the API works, just more slowly, when Redis is down

The Real-time Notifications mini-project (capsule 05)

A complete system: a publisher (the API) + a subscriber (a worker)
Pub/Sub for cache invalidation events
A WebSocket bridge: subscribers receive events from Redis and broadcast them to WebSocket clients
Connection pooling configured correctly
Integration tests

Connection with the capstone project

Module 5's project (Production Cached API) uses everything from this module:

Feature of the capstone project	Its module 4 origin
Async Redis throughout the stack	Capsules 03-04
A connection pool with max_connections	Capsule 03
`Depends(get_redis)` in the endpoints	Capsule 04
Cache invalidation events via Pub/Sub	Capsules 02, 04
Automatic caching middleware	Capsule 04
A health check endpoint with a Redis check	Capsule 04
Graceful degradation with logging	Capsule 04

This module gives you the professional scaffolding on top of which M5 builds the complete app.

The module's 5 capsules

Capsule 01: Module introduction (you are here)
    → Sync vs async, communication between components, a preview

Capsule 02: Pub/Sub
    → PUBLISH, SUBSCRIBE, PSUBSCRIBE, use cases, limitations

Capsule 03: redis.asyncio + Connection Pooling
    → The official async client, ConnectionPool, the singleton pattern

Capsule 04: FastAPI Integration
    → Dependency injection, lifespan, caching middleware, health checks, graceful degradation

Capsule 05: The Real-time Notifications mini-project
    → A publisher + subscriber, a WebSocket bridge, integration tests

Estimated time: 2.0-2.5 hours of active study.

The module's critical rule

If you're coming from old tutorials (pre-2022), you've probably seen code like:

# ❌ DEPRECATED since 2021
import aioredis

async def main():
    r = await aioredis.create_redis_pool('redis://localhost')
    await r.set("key", "value")

This is no longer used. The aioredis library was absorbed into redis-py in version 4.2 (2022). The modern client is:

# ✅ MODERN (redis-py 4.2+, 5.x)
import redis.asyncio as aioredis  # an alias by convention

async def main():
    r = aioredis.Redis(host='localhost', port=6379, decode_responses=True)
    await r.set("key", "value")

All you need is pip install redis — the redis.asyncio module is included. If you install aioredis separately, you'll get dependency conflicts and code that's no longer maintained.

This is the same situation as the Auth guide with passlib (deprecated) → pwdlib (modern). A verified 2026 stack.

What you will NOT learn in this module

These topics are intentionally left out:

❌ Redis Streams (the modern alternative to Pub/Sub with persistence) — an advanced feature, out of scope
❌ Real message brokers: RabbitMQ, Kafka, NATS — they're separate tools, not Redis
❌ Celery with a Redis broker — a specific companion guide
❌ Bidirectional WebSockets with full auth — guide #7 (FastAPI Advanced) covers it
❌ Server-Sent Events (SSE) — an alternative to WebSockets, we don't cover it
❌ Distributed Pub/Sub across DCs — Redis Cluster or more advanced solutions

If you need any of these later, the resources at the end point to references.

Prerequisites

Software (you should have this from modules 1-3)

✅ Redis running in Docker
✅ Python 3.10+ with a virtual environment
✅ redis-py ≥ 4.2 installed (pip show redis should ideally show version 5.x)

Creating the module 4 workspace

mkdir -p ~/projects/redis-guide/module-04-async-pubsub
cd ~/projects/redis-guide/module-04-async-pubsub
python -m venv .venv
source .venv/bin/activate
pip install "redis>=7.4" fastapi uvicorn[standard] httpx pytest pytest-asyncio

Verifying the correct version

python -c "import redis; print('redis-py version:', redis.__version__)"
# it should show: redis-py version: 5.x.x

If you have < 4.2, upgrade with pip install --upgrade redis.

Prior knowledge

From module 1:

✅ Synchronous redis-py: set(), get(), hset(), zadd(), etc.

From module 2:

✅ Cache invalidation (you'll know WHY Pub/Sub solves it better than a manual DELETE)

From module 3:

✅ FastAPI middleware (you'll extend the pattern here)

From the path:

✅ Async/await in Python (a basic sense of how it works)
✅ FastAPI dependency injection (Guide #7)

⚠️ If async/await feels rusty: take a quick look at how it works before continuing. A good resource: Real Python: Async IO in Python.

The module's teaching philosophy

Three principles:

1. Async isn't "more advanced" — it's the correct way to do I/O in FastAPI

Some guides present async as an optional feature for "heavily scaled" apps. That's misleading. FastAPI IS async. Using sync inside async endpoints is an anti-pattern that kills performance. It isn't optional for production.

2. Pub/Sub for what it's good at — not for what it isn't

Redis Pub/Sub is excellent for cache invalidation and lightweight notifications. It's TERRIBLE for durable messaging (there's no persistence, messages get lost). You'll understand both sides clearly — we don't sell Pub/Sub as a universal solution.

3. Connection pooling is hygiene, not optimization

Without pooling, every request opens a new connection. With 1000 concurrent requests, you open 1000 connections. Redis has a default limit of 10,000 connections — you'll blow past it fast. Pooling isn't an advanced feature — it's like closing file handles. You always do it.

Resources to get started

Official documentation

redis.asyncio API Reference — The official async client
redis-py Migration Guide — From aioredis (deprecated) to redis.asyncio
Redis Pub/Sub Documentation — The official Pub/Sub docs with the command syntax
FastAPI Dependencies — The DI patterns we'll use

To understand it from another angle

Async IO in Python — A deep async/await tutorial
Pub/Sub vs Message Queue — A comparison of the paradigms

Before moving on to capsule 02

Take 5 minutes:

Verify your redis-py version:

python -c "import redis; print(redis.__version__)"
# 5.x.x ✓

Create the workspace:

mkdir -p ~/projects/redis-guide/module-04-async-pubsub
cd ~/projects/redis-guide/module-04-async-pubsub
python -m venv .venv
source .venv/bin/activate
pip install "redis>=7.4" fastapi uvicorn[standard] httpx

Verify Redis is up:

docker ps | grep redis
redis-cli ping  # PONG

A quick test of the async client:

python -c "
import asyncio
from redis.asyncio import Redis

async def main():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    await r.set('test', 'async ok')
    v = await r.get('test')
    print(v)
    await r.close()

asyncio.run(main())
"
# Output: async ok

If all 4 are ✅, you're ready.

Summary

In this capsule you understood:

Sync inside async endpoints is an anti-pattern. It blocks the event loop and kills performance under concurrency
redis.asyncio is the modern client — it comes with pip install redis ≥ 4.2. Do NOT use the deprecated aioredis
Pub/Sub solves communication between components — decoupled, scalable, simple
Pub/Sub has real limitations: fire-and-forget, no persistence, no replay. For durable messaging, RabbitMQ/Kafka
Connection pooling isn't optional — 1000 connections for 1000 requests blows up Redis
Professional FastAPI integration: DI with Depends(get_redis), lifespan events, caching middleware, health checks
A verified 2026 stack: redis>=7.4, redis.asyncio (not aioredis)

In capsule 02 you get into Pub/Sub: how to publish events, how to subscribe, what you can do with it (cache invalidation, notifications) and what you can NOT (durable messaging).

What's next?

Capsule 02: Pub/Sub — PUBLISH, SUBSCRIBE, PSUBSCRIBE with pattern matching, real use cases (event-driven cache invalidation, notifications), honest limitations (fire-and-forget), and when to use another tool instead.

If your workspace is ready and redis.asyncio works, let's go.

← PreviousMini-project: Rate Limiter Service Next →Pub/Sub