Module 4: Pub/Sub and FastAPI Integration

Mini-project: Real-time Notifications

Overview

You close module 4 with a project that combines the 4 patterns you learned: redis.asyncio, connection pooling, Pub/Sub, and professional FastAPI integration. The Real-time Notifications Service is a FastAPI app that receives events via HTTP POST, publishes them to Redis Pub/Sub, and broadcasts them to connected clients via WebSocket. It's exactly the pattern chat apps, collaborative editing, live dashboards, and real-time notifications use.

It isn't a step-by-step tutorial — it's clear specs and reference code. You can implement it from scratch or copy and understand it. What matters is seeing how the pieces fit together: an HTTP endpoint that publishes → Pub/Sub that distributes → a WebSocket that delivers. You've seen every piece in earlier capsules; here you integrate them into a coherent app. After module 5 you'll have two portfolio-worthy projects: this one (Pub/Sub + a WebSocket bridge) and the Production Cached API. Both demonstrate command of advanced Redis, not just "I use a cache."

The project includes integration tests with pytest-asyncio that verify the end-to-end flow: send a POST, wait for the WebSocket message, validate that it arrived. It's portfolio-worthy: the tests show you understand async + WebSockets + Redis Pub/Sub working together. By the end of the capsule, you've closed module 4 at 100% and only module 5 remains to finish the guide.

The project's specs

Functionality

1. A WebSocket client connects to /ws/{topic}
2. The WebSocket bridge subscribes Redis to the pattern "topic:{topic}:*"
3. An HTTP POST to /events/{topic}/{event_type} publishes an event
4. Pub/Sub distributes the event
5. The bridge receives it from the subscriber and broadcasts it to ALL the topic's WebSocket clients

Use cases

Chat: topic = chat_room_id, events = messages
Collaborative editing: topic = doc_id, events = document changes
Live dashboards: topic = "metrics", events = new metrics
Notifications: topic = user_id, events = personal notifications

Endpoints

GET  /                               → A welcome page with a WebSocket demo
WS   /ws/{topic}                     → A WebSocket connected to a topic
POST /events/{topic}/{event_type}    → Publish an event (admin/internal)
GET  /topics/{topic}/clients          → How many clients are connected to a topic
GET  /health                          → Health check

Project structure

realtime-notifications/
├── .venv/
├── app/
│   ├── __init__.py
│   ├── main.py              # The FastAPI app + lifespan
│   ├── config.py
│   ├── redis_client.py      # The pool's singleton
│   ├── ws_manager.py        # The connection manager for WebSockets
│   └── pubsub_listener.py   # The Redis Pub/Sub subscriber
├── tests/
│   ├── __init__.py
│   └── test_integration.py
├── static/
│   └── index.html           # The demo client
├── requirements.txt
└── README.md

`requirements.txt`

fastapi>=0.136
uvicorn[standard]>=0.27.0
redis>=7.4
pydantic>=2.0
httpx>=0.26.0
pytest>=8.0.0
pytest-asyncio>=0.23.0
websockets>=12.0

`app/config.py`

"""Configuration."""
import os


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

# The Pub/Sub channel pattern
CHANNEL_PATTERN = "notifications:*"
CHANNEL_PREFIX = "notifications"

`app/redis_client.py`

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

from app.config import REDIS_URL, REDIS_MAX_CONNECTIONS


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


def init_pool():
    global _pool, _client
    _pool = ConnectionPool.from_url(
        REDIS_URL,
        max_connections=REDIS_MAX_CONNECTIONS,
        decode_responses=True,
        socket_timeout=2.0,
        socket_keepalive=True,
        health_check_interval=30,
    )
    _client = Redis(connection_pool=_pool)


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


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

`app/ws_manager.py`

"""
The connection manager: it tracks WebSocket connections per topic.
"""
import logging
from collections import defaultdict
from fastapi import WebSocket
from typing import Set


logger = logging.getLogger(__name__)


class ConnectionManager:
    """Manages WebSocket connections grouped by topic."""

    def __init__(self):
        # topic -> set of WebSocket connections
        self.connections: dict[str, Set[WebSocket]] = defaultdict(set)

    async def connect(self, ws: WebSocket, topic: str):
        await ws.accept()
        self.connections[topic].add(ws)
        logger.info(f"WS connected to topic '{topic}'. Total in topic: {len(self.connections[topic])}")

    def disconnect(self, ws: WebSocket, topic: str):
        if ws in self.connections[topic]:
            self.connections[topic].remove(ws)
        if not self.connections[topic]:
            del self.connections[topic]
        logger.info(f"WS disconnected from topic '{topic}'")

    async def broadcast(self, topic: str, message: dict):
        """Send message to all connections in a topic."""
        if topic not in self.connections:
            return 0

        # Copy to avoid mutation during iteration
        recipients = list(self.connections[topic])
        sent = 0
        disconnected = []

        for ws in recipients:
            try:
                await ws.send_json(message)
                sent += 1
            except Exception as e:
                logger.warning(f"Failed to send to WS: {e}. Marking for disconnect.")
                disconnected.append(ws)

        # Clean up dead connections
        for ws in disconnected:
            self.disconnect(ws, topic)

        return sent

    def total_clients(self) -> int:
        return sum(len(s) for s in self.connections.values())

    def clients_in_topic(self, topic: str) -> int:
        return len(self.connections.get(topic, set()))


# Singleton
ws_manager = ConnectionManager()

`app/pubsub_listener.py`

"""
A background task: it listens to Redis Pub/Sub and broadcasts to WebSocket clients.
"""
import asyncio
import json
import logging
from redis.asyncio import Redis
from redis.exceptions import RedisError

from app.config import CHANNEL_PATTERN, CHANNEL_PREFIX
from app.redis_client import get_redis
from app.ws_manager import ws_manager


logger = logging.getLogger(__name__)


async def listen_and_broadcast():
    """
    A background task: subscribe to Redis Pub/Sub and forward to WebSocket clients.
    It auto-reconnects if Redis fails.
    """
    while True:
        try:
            r = get_redis()
            pubsub = r.pubsub()
            await pubsub.psubscribe(CHANNEL_PATTERN)
            logger.info(f"Pub/Sub listener subscribed to '{CHANNEL_PATTERN}'")

            async for message in pubsub.listen():
                if message["type"] != "pmessage":
                    continue

                # Parse the channel: "notifications:<topic>:<event_type>"
                channel = message["channel"]
                parts = channel.split(":", 2)
                if len(parts) < 3:
                    continue

                _, topic, event_type = parts

                # Parse the data
                try:
                    data = json.loads(message["data"])
                except json.JSONDecodeError:
                    data = message["data"]

                # Broadcast to WebSocket clients in this topic
                event = {
                    "type": event_type,
                    "topic": topic,
                    "data": data,
                }
                sent = await ws_manager.broadcast(topic, event)

                if sent > 0:
                    logger.debug(f"Broadcasted '{event_type}' to {sent} clients in '{topic}'")

        except (RedisError, ConnectionError) as e:
            logger.error(f"Pub/Sub error: {e}. Reconnecting in 5s...")
            await asyncio.sleep(5)
        except Exception as e:
            logger.error(f"Unexpected error: {e}", exc_info=True)
            await asyncio.sleep(5)

`app/main.py` (the main FastAPI app)

"""
Real-time Notifications Service.
"""
import asyncio
import json
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, Depends
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from redis.asyncio import Redis

from app.config import CHANNEL_PREFIX
from app.redis_client import init_pool, close_pool, get_redis
from app.ws_manager import ws_manager
from app.pubsub_listener import listen_and_broadcast


logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s: %(message)s")
logger = logging.getLogger(__name__)


# ═══════════════════════════════════════════════════════════
# Lifespan
# ═══════════════════════════════════════════════════════════


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    init_pool()

    r = get_redis()
    try:
        await r.ping()
        logger.info("✓ Redis connected")
    except Exception as e:
        logger.error(f"✗ Redis failed: {e}")
        raise

    # Start the background listener
    listener_task = asyncio.create_task(listen_and_broadcast())
    logger.info("✓ Pub/Sub listener started")

    yield

    # Shutdown
    listener_task.cancel()
    try:
        await listener_task
    except asyncio.CancelledError:
        pass
    await close_pool()
    logger.info("✓ Cleanup complete")


# ═══════════════════════════════════════════════════════════
# App + the dependency
# ═══════════════════════════════════════════════════════════


app = FastAPI(
    title="Real-time Notifications Service",
    description="WebSocket bridge for Redis Pub/Sub events",
    version="1.0.0",
    lifespan=lifespan,
)


async def get_redis_dep() -> Redis:
    return get_redis()


# ═══════════════════════════════════════════════════════════
# The WebSocket endpoint
# ═══════════════════════════════════════════════════════════


@app.websocket("/ws/{topic}")
async def websocket_endpoint(websocket: WebSocket, topic: str):
    """
    WebSocket: the client connects and receives the topic's events.
    """
    await ws_manager.connect(websocket, topic)

    # Send welcome
    await websocket.send_json({
        "type": "connected",
        "topic": topic,
        "message": f"Connected to topic '{topic}'",
    })

    try:
        while True:
            # Keep the connection open. If the client sends something, we ignore it
            # (this service only broadcasts, it doesn't process client input)
            data = await websocket.receive_text()
            # Echo back as ping
            await websocket.send_json({"type": "echo", "received": data})
    except WebSocketDisconnect:
        ws_manager.disconnect(websocket, topic)
    except Exception as e:
        logger.warning(f"WS error in topic '{topic}': {e}")
        ws_manager.disconnect(websocket, topic)


# ═══════════════════════════════════════════════════════════
# HTTP endpoints
# ═══════════════════════════════════════════════════════════


class EventData(BaseModel):
    payload: dict | None = None
    metadata: dict | None = None


@app.post("/events/{topic}/{event_type}")
async def publish_event(
    topic: str,
    event_type: str,
    body: EventData,
    r: Redis = Depends(get_redis_dep),
):
    """
    Publish an event. It reaches the topic's WebSocket clients via Pub/Sub.
    """
    channel = f"{CHANNEL_PREFIX}:{topic}:{event_type}"

    payload = {
        "payload": body.payload or {},
        "metadata": body.metadata or {},
    }

    try:
        subscribers = await r.publish(channel, json.dumps(payload))
        logger.info(f"Published to {channel}. {subscribers} pub/sub subscribers (incl. our bridge).")

        # How many WebSocket clients are in this topic
        ws_count = ws_manager.clients_in_topic(topic)

        return {
            "published": True,
            "channel": channel,
            "pubsub_subscribers": subscribers,
            "ws_clients_in_topic": ws_count,
        }
    except Exception as e:
        logger.error(f"Publish failed: {e}")
        raise HTTPException(503, f"Publish failed: {e}")


@app.get("/topics/{topic}/clients")
def topic_clients(topic: str):
    """How many WebSocket clients are connected to a topic."""
    return {
        "topic": topic,
        "ws_clients": ws_manager.clients_in_topic(topic),
        "total_ws_clients": ws_manager.total_clients(),
    }


@app.get("/health")
async def health(r: Redis = Depends(get_redis_dep)):
    try:
        await r.ping()
        return {
            "status": "healthy",
            "ws_clients_total": ws_manager.total_clients(),
            "ws_topics_active": len(ws_manager.connections),
        }
    except Exception as e:
        return {
            "status": "degraded",
            "redis_error": str(e),
            "message": "Pub/Sub bridge unavailable. WebSockets won't receive events.",
        }


# ═══════════════════════════════════════════════════════════
# The demo HTML
# ═══════════════════════════════════════════════════════════


@app.get("/", response_class=HTMLResponse)
def demo():
    return """
<!DOCTYPE html>
<html>
<head><title>Real-time Notifications Demo</title></head>
<body>
    <h1>Real-time Notifications Demo</h1>
    <input id="topic" placeholder="topic" value="news">
    <button onclick="connect()">Connect</button>
    <button onclick="disconnect()">Disconnect</button>
    <pre id="log"></pre>

    <h3>Send event (server side)</h3>
    <pre>
    curl -X POST http://localhost:8000/events/news/update \\
      -H "Content-Type: application/json" \\
      -d '{"payload": {"title": "Breaking news"}}'
    </pre>

    <script>
        let ws = null;
        const log = (msg) => {
            const el = document.getElementById("log");
            el.textContent += new Date().toISOString() + ": " + msg + "\\n";
        };

        function connect() {
            const topic = document.getElementById("topic").value;
            ws = new WebSocket(`ws://localhost:8000/ws/${topic}`);
            ws.onopen = () => log("Connected to topic: " + topic);
            ws.onmessage = (e) => log("Received: " + e.data);
            ws.onclose = () => log("Disconnected");
            ws.onerror = (e) => log("Error: " + e);
        }

        function disconnect() {
            if (ws) ws.close();
        }
    </script>
</body>
</html>
    """

Running it

# Setup
cd ~/projects/redis-guide/module-04-async-pubsub
mkdir -p realtime-notifications && cd realtime-notifications
mkdir -p app tests static

# Copy the project's files

# Set up the venv
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

# Verify Redis
docker ps | grep redis

# Start it
uvicorn app.main:app --reload --port 8000

Expected output:

INFO:     Uvicorn running on http://127.0.0.1:8000
2026-04-25 10:00:00 [__main__] INFO: ✓ Redis connected
2026-04-25 10:00:00 [__main__] INFO: ✓ Pub/Sub listener started
INFO:     Application startup complete.

Test 1: WebSocket + HTTP publish (manual)

The browser test

Open http://localhost:8000/ in the browser
Click "Connect" (topic: news)
You'll see: Connected to topic: news

In another terminal, publish an event:

curl -X POST http://localhost:8000/events/news/update \
  -H "Content-Type: application/json" \
  -d '{"payload": {"title": "Breaking news", "author": "Editor"}}'

In the browser you'll see:

Received: {"type": "update", "topic": "news", "data": {"payload": {"title": "Breaking news", "author": "Editor"}, "metadata": {}}}

A CLI test with websocat

# Terminal 1: the WebSocket subscriber
websocat ws://localhost:8000/ws/news

# Terminal 2: publish
curl -X POST http://localhost:8000/events/news/update \
  -H "Content-Type: application/json" \
  -d '{"payload": {"title": "From CLI"}}'

Terminal 1 shows the event as JSON.

Multiple clients

# Terminal 1
websocat ws://localhost:8000/ws/news

# Terminal 2
websocat ws://localhost:8000/ws/news

# Terminal 3
websocat ws://localhost:8000/ws/sports

# Check the clients per topic
curl http://localhost:8000/topics/news/clients
# {"topic":"news","ws_clients":2,"total_ws_clients":3}

# Publish to news → terminals 1 and 2 receive it, NOT terminal 3
curl -X POST http://localhost:8000/events/news/update -H "Content-Type: application/json" -d '{"payload": {"x":1}}'

# Publish to sports → only terminal 3
curl -X POST http://localhost:8000/events/sports/score -H "Content-Type: application/json" -d '{"payload": {"y":2}}'

Test 2: Integration tests with pytest

Create tests/test_integration.py:

"""
Integration tests: end-to-end of the Pub/Sub → WebSocket flow.
"""
import asyncio
import json
import pytest
import httpx
from contextlib import asynccontextmanager


BASE_URL = "http://localhost:8000"
WS_URL = "ws://localhost:8000"


@pytest.mark.asyncio
async def test_health():
    """Verify the service is running."""
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{BASE_URL}/health")
        assert r.status_code == 200
        data = r.json()
        assert data["status"] in ("healthy", "degraded")


@pytest.mark.asyncio
async def test_publish_to_no_subscribers():
    """Publishing without subscribers should still succeed (Pub/Sub fire-and-forget)."""
    async with httpx.AsyncClient() as client:
        r = await client.post(
            f"{BASE_URL}/events/test_no_subs/event_type",
            json={"payload": {"test": "data"}},
        )
        assert r.status_code == 200
        data = r.json()
        assert data["published"] is True
        # ws_clients_in_topic might be 0 (no one subscribed)
        assert "pubsub_subscribers" in data


@pytest.mark.asyncio
async def test_websocket_receives_event():
    """End-to-end: WebSocket connects, HTTP publishes, WebSocket receives."""
    import websockets

    topic = "test_e2e"

    async with websockets.connect(f"{WS_URL}/ws/{topic}") as ws:
        # Receive welcome
        welcome = await ws.recv()
        welcome_data = json.loads(welcome)
        assert welcome_data["type"] == "connected"
        assert welcome_data["topic"] == topic

        # A small delay so the bridge subscribes (race condition tolerance)
        await asyncio.sleep(0.1)

        # Publish the event via HTTP
        async with httpx.AsyncClient() as client:
            r = await client.post(
                f"{BASE_URL}/events/{topic}/test_event",
                json={"payload": {"hello": "world"}},
            )
            assert r.status_code == 200

        # Receive event via WebSocket
        message = await asyncio.wait_for(ws.recv(), timeout=2.0)
        msg_data = json.loads(message)

        assert msg_data["type"] == "test_event"
        assert msg_data["topic"] == topic
        assert msg_data["data"]["payload"]["hello"] == "world"


@pytest.mark.asyncio
async def test_multiple_clients_same_topic():
    """Multiple WebSocket clients in the same topic should ALL receive events."""
    import websockets

    topic = "test_broadcast"

    async with websockets.connect(f"{WS_URL}/ws/{topic}") as ws1:
        async with websockets.connect(f"{WS_URL}/ws/{topic}") as ws2:
            # Skip welcomes
            await ws1.recv()
            await ws2.recv()

            # Publish
            await asyncio.sleep(0.1)
            async with httpx.AsyncClient() as client:
                await client.post(
                    f"{BASE_URL}/events/{topic}/broadcast_test",
                    json={"payload": {"to": "all"}},
                )

            # Both should receive
            msg1 = json.loads(await asyncio.wait_for(ws1.recv(), timeout=2.0))
            msg2 = json.loads(await asyncio.wait_for(ws2.recv(), timeout=2.0))

            assert msg1["data"]["payload"]["to"] == "all"
            assert msg2["data"]["payload"]["to"] == "all"


@pytest.mark.asyncio
async def test_topic_isolation():
    """Clients in different topics should NOT receive each other's events."""
    import websockets

    async with websockets.connect(f"{WS_URL}/ws/topic_a") as ws_a:
        async with websockets.connect(f"{WS_URL}/ws/topic_b") as ws_b:
            await ws_a.recv()  # welcome
            await ws_b.recv()  # welcome

            # Publish to topic_a only
            await asyncio.sleep(0.1)
            async with httpx.AsyncClient() as client:
                await client.post(
                    f"{BASE_URL}/events/topic_a/event",
                    json={"payload": {"only_a": True}},
                )

            # ws_a should receive
            msg_a = json.loads(await asyncio.wait_for(ws_a.recv(), timeout=2.0))
            assert msg_a["data"]["payload"]["only_a"] is True

            # ws_b should NOT receive (timeout)
            with pytest.raises(asyncio.TimeoutError):
                await asyncio.wait_for(ws_b.recv(), timeout=1.0)


@pytest.mark.asyncio
async def test_topic_clients_count():
    """Verify the /topics/{topic}/clients endpoint."""
    import websockets

    topic = "test_count"

    # Initially 0
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{BASE_URL}/topics/{topic}/clients")
        assert r.json()["ws_clients"] == 0

    # Connect one
    async with websockets.connect(f"{WS_URL}/ws/{topic}") as ws:
        await ws.recv()  # welcome

        await asyncio.sleep(0.1)  # let manager update

        async with httpx.AsyncClient() as client:
            r = await client.get(f"{BASE_URL}/topics/{topic}/clients")
            assert r.json()["ws_clients"] == 1

    # After disconnect: back to 0
    await asyncio.sleep(0.5)
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{BASE_URL}/topics/{topic}/clients")
        assert r.json()["ws_clients"] == 0

Running the tests

# Make sure the server is running
# In another terminal:
pytest tests/test_integration.py -v -s

Expected output:

tests/test_integration.py::test_health PASSED
tests/test_integration.py::test_publish_to_no_subscribers PASSED
tests/test_integration.py::test_websocket_receives_event PASSED
tests/test_integration.py::test_multiple_clients_same_topic PASSED
tests/test_integration.py::test_topic_isolation PASSED
tests/test_integration.py::test_topic_clients_count PASSED

==================== 6 passed in 4.2s ====================

✅ The end-to-end flow works correctly.

A real use case: cache invalidation events

This service isn't just for notifications. It's exactly the pattern for cache invalidation events in a distributed app.

# admin_service.py
async def update_product(product_id, data):
    db.update(product_id, data)

    # Notify all subscribers via this service
    async with httpx.AsyncClient() as client:
        await client.post(
            f"http://realtime-svc.internal:8000/events/cache/invalidate",
            json={
                "payload": {"product_id": product_id, "reason": "admin_update"},
            }
        )

# api_service.py (a subscriber via WebSocket)
async def listen_for_invalidations():
    async with websockets.connect("ws://realtime-svc.internal:8000/ws/cache") as ws:
        async for raw_message in ws:
            event = json.loads(raw_message)
            product_id = event["data"]["payload"]["product_id"]

            # Invalidate local caches
            await r.delete(f"cache:product:{product_id}")
            await r.delete("cache:products:list")

The service you built is reusable — it works for chat, cache invalidation, dashboards, notifications, whatever.

A professional README

Create README.md:

# Real-time Notifications Service

WebSocket bridge for Redis Pub/Sub. Publish events via HTTP, deliver to WebSocket clients in real-time.

## Features

- 🚀 **Sub-millisecond latency** — Pub/Sub + WebSocket
- 🎯 **Topic-based** — Clients subscribe to topics, events route automatically
- 🔄 **Auto-reconnect** — Subscriber reconnects to Redis on failure
- 📊 **Health checks** — Monitor service and Redis status
- 🧪 **Tested** — Integration tests with pytest-asyncio

## Quick Start

```bash
docker run -d --name redis-dev -p 6379:6379 redis:7

git clone <repo>
cd realtime-notifications
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

uvicorn app.main:app --reload

Open http://localhost:8000/ for the demo.

Use Cases

Use case	Topic	Event types
Chat	`room:{id}`	`message`, `typing`, `read`
Collaborative editing	`doc:{id}`	`edit`, `cursor`, `selection`
Dashboard updates	`metrics`	`update`, `alert`
Cache invalidation	`cache`	`invalidate`, `refresh`
User notifications	`user:{id}`	`notification`, `badge`

API

WebSocket: subscribe

ws://host/ws/{topic}

HTTP: publish

POST /events/{topic}/{event_type}
Body: {"payload": {...}, "metadata": {...}}

Status

GET /topics/{topic}/clients   # How many WS clients in topic
GET /health                    # Service + Redis status

Testing

pytest tests/ -v

Architecture

HTTP POST → Redis PUBLISH → Subscriber Bridge → WebSocket Broadcast

Limitations

Pub/Sub fire-and-forget: if no clients are connected, events are lost
No persistence: for durable messaging, use Redis Streams or RabbitMQ
Single-region: for multi-DC, consider Kafka or specialized pub/sub services

License

MIT


---

## Troubleshooting

### Problem 1: The tests time out on `await ws.recv()`

**Cause:** The bridge listener isn't processing messages fast enough, or the subscriber isn't connected.

**Solution:**
1. Check the server's logs — does `✓ Pub/Sub listener started` show up?
2. Raise the `await asyncio.sleep(0.1)` before the publish to 0.5s
3. Verify you're checking `pmessage` (not `message`) in the listener

### Problem 2: WebSocket clients don't receive published events

**Cause:** The listener failed or never started.

**Solution:**
- Check the server's logs
- Verify `await pubsub.psubscribe(CHANNEL_PATTERN)` is in the listener
- The `CHANNEL_PATTERN` should match `CHANNEL_PREFIX:*`

### Problem 3: A WebSocket disconnect causes errors in the logs

**Cause:** The client closes the connection and the manager tries to send.

**Solution:** Already implemented — `broadcast()` catches exceptions and cleans up dead connections.

### Problem 4: The HTTP publish returns a 503 when Redis fails

**Cause:** The service is degraded and can't publish.

**Solution:** This is correct — if Redis is down, the service CAN'T do its job. Returning a 503 is honest. The client should retry.

### Problem 5: Memory grows with clients that don't close cleanly

**Cause:** Zombie connections piling up.

**Solution:** `broadcast()` cleans up clients that fail to receive. The timing: roughly whenever an event arrives for the topic. For more aggressive cleanup, add a periodic task that checks connections with `ws.ping()`.

---

## Module 4 summary

You close module 4 with command of:

**`redis.asyncio`:**
- The modern client (`redis-py >= 4.2`)
- An API identical to the sync one with `await`
- `aclose()` for clean shutdown
- Do NOT use `aioredis` (deprecated since 2021)

**Pub/Sub:**
- `PUBLISH`, `SUBSCRIBE`, `PSUBSCRIBE` with wildcards
- Use cases: cache invalidation events, notifications, broadcasting
- Limitations: fire-and-forget, no persistence, no replay
- When to use Pub/Sub vs Redis Streams vs message brokers

**Connection Pooling:**
- A `ConnectionPool` with a tunable `max_connections`
- The singleton pattern to share it across the whole app
- Health checks with `health_check_interval`
- Reconnection with `Retry` + `ExponentialBackoff`

**FastAPI Integration:**
- Lifespan events (modern, replacing `@app.on_event`)
- Dependency injection with `Depends(get_redis)`
- Automatic caching middleware
- Health checks with a Redis check
- Graceful degradation when Redis fails

**The Real-time Notifications mini-project:**
- A WebSocket bridge for Redis Pub/Sub
- An HTTP publish + a WebSocket subscribe
- A connection manager per topic
- A background listener with auto-reconnect
- End-to-end integration tests

**What's coming in module 5:** The final capstone project — the **Production Cached API**. You assemble EVERYTHING from the 4 modules: a complete caching strategy per endpoint, multi-tier rate limiting, sessions with JWT, Pub/Sub for invalidation events, monitoring with metrics, Docker Compose. It's the portfolio piece that demonstrates total command of advanced Redis.

---

## Additional resources

1. [FastAPI WebSockets](https://fastapi.tiangolo.com/advanced/websockets/) — The official docs
2. [websockets library](https://websockets.readthedocs.io/) — The Python client for tests
3. [pytest-asyncio](https://pytest-asyncio.readthedocs.io/) — Testing async
4. [Real-time Web Apps Architecture](https://ably.com/topic/architecture-patterns-of-realtime-applications) — General patterns
5. [Redis Pub/Sub Best Practices](https://redis.com/blog/communicate-using-redis-pub-sub-pattern/) — Real cases
6. [Building Scalable WebSocket Services](https://www.ably.com/blog/scaling-websockets) — When a WebSocket bridge isn't enough

---

## What's next?

You've closed **Module 4: Pub/Sub & FastAPI Integration**. In **Module 5: Capstone Project — Production Cached API** you get to the end of the guide: assembling ALL the modules into a portfolio-worthy API.

Before moving on, make sure you:

- [ ] Have Real-time Notifications running
- [ ] Have the integration tests passing (6/6)
- [ ] Have the demo HTML working (the browser can receive events)
- [ ] Understand the flow: HTTP POST → Pub/Sub → WebSocket
- [ ] Have clean logs with no errors

If all 5 are ✅, you've completed module 4. On to module 5 — the last one.

← PreviousDeep FastAPI Integration Next →Final Project: Production Cached API