Tech Tutorial - February 18 2026 233008
Welcome to today’s deep‑dive tutorial! In this guide we’ll walk through building a production‑ready real‑time chat service using FastAPI, WebSockets, and Redis. By the end you’ll have a solid grasp of asynchronous programming in Python, understand how to scale WebSocket connections, and be ready to deploy the solution on a cloud provider. Grab your favorite editor, and let’s start turning concepts into working code.
Why FastAPI and WebSockets?
FastAPI has become the go‑to framework for modern Python APIs thanks to its speed, type safety, and native async support. When you pair it with WebSockets you unlock full‑duplex communication—perfect for chat, live dashboards, or multiplayer games. Unlike traditional HTTP polling, WebSockets keep a single TCP connection open, dramatically reducing latency and bandwidth overhead.
Redis enters the picture as an in‑memory data store that can act as a message broker. Using Redis Pub/Sub we can broadcast messages to every connected client without coupling the WebSocket handlers directly. This architecture scales horizontally: spin up more FastAPI workers, and they’ll all listen to the same Redis channel.
Project Layout
Before we write code, let’s outline the directory structure. Keeping files organized makes future maintenance a breeze.
chat_app/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI entry point
│ ├── router.py # WebSocket routes
│ └── utils.py # Helper functions (e.g., message validation)
├── requirements.txt
└── Dockerfile
We’ll focus on main.py, router.py, and utils.py. The Dockerfile is optional but included for a one‑click deployment experience.
Setting Up the Environment
First, create a virtual environment and install the dependencies.
python -m venv venv
source venv/bin/activate
pip install fastapi[all] uvicorn redis
FastAPI[all] pulls in pydantic, starlette, and uvicorn for ASGI serving. The redis package gives us a simple async client.
Creating the FastAPI App
Open app/main.py and bootstrap the application.
from fastapi import FastAPI
from .router import router
app = FastAPI(
title="Real‑Time Chat Service",
description="A FastAPI + WebSocket + Redis demo",
version="0.1.0",
)
app.include_router(router)
Notice the separation of concerns: routing logic lives in router.py, keeping main.py clean and testable.
WebSocket Endpoint & Redis Pub/Sub
Now let’s implement the core WebSocket logic. Open app/router.py and add the following:
import json
import uuid
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from redis.asyncio import Redis
router = APIRouter()
redis = Redis(host="redis", port=6379, decode_responses=True)
# Helper to broadcast a message to all subscribers
async def broadcast(channel: str, message: dict):
await redis.publish(channel, json.dumps(message))
@router.websocket("/ws/chat/{room_id}")
async def chat_endpoint(websocket: WebSocket, room_id: str):
await websocket.accept()
client_id = str(uuid.uuid4())
pubsub = redis.pubsub()
await pubsub.subscribe(room_id)
async def send_messages():
async for msg in pubsub.listen():
if msg["type"] == "message":
await websocket.send_text(msg["data"])
# Run the listener in the background
listener_task = asyncio.create_task(send_messages())
try:
while True:
data = await websocket.receive_text()
payload = {
"client_id": client_id,
"room_id": room_id,
"message": data,
"timestamp": datetime.utcnow().isoformat()
}
await broadcast(room_id, payload)
except WebSocketDisconnect:
await pubsub.unsubscribe(room_id)
listener_task.cancel()
This endpoint does three things:
- Accepts a WebSocket connection and assigns a unique
client_id. - Subscribes the connection to a Redis channel named after the
room_id. - Listens for incoming messages, wraps them in a JSON payload, and publishes them back to Redis for all subscribers.
Because Redis handles the fan‑out, each FastAPI worker can stay lightweight, merely relaying messages.
Pro tip: Use redis.asyncio (available since redis‑py 4.2) to avoid blocking the event loop. Mixing sync Redis calls with async WebSockets will cause performance bottlenecks.
Message Validation with Pydantic
Before broadcasting, it’s wise to validate the payload. Create app/utils.py:
from pydantic import BaseModel, Field, validator
from datetime import datetime
class ChatMessage(BaseModel):
client_id: str = Field(..., description="Unique identifier for the sender")
room_id: str = Field(..., description="Chat room identifier")
message: str = Field(..., min_length=1, max_length=500)
timestamp: datetime
@validator("message")
def no_prohibited_words(cls, v):
prohibited = {"spam", "advertisement"}
if any(word in v.lower() for word in prohibited):
raise ValueError("Message contains prohibited content")
return v
Now modify router.py to use the model:
from .utils import ChatMessage
# Inside chat_endpoint loop
while True:
raw = await websocket.receive_text()
try:
payload = ChatMessage(
client_id=client_id,
room_id=room_id,
message=raw,
timestamp=datetime.utcnow()
)
except ValueError as e:
await websocket.send_text(json.dumps({"error": str(e)}))
continue
await broadcast(room_id, payload.dict())
With validation in place, you protect your service from malformed or malicious payloads without sacrificing performance.
Running Locally
Spin up a Redis container and the FastAPI server with Docker Compose. Create a docker-compose.yml in the project root:
version: "3.9"
services:
redis:
image: redis:7-alpine
ports:
- "6379:6379"
api:
build: .
command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
volumes:
- .:/code
ports:
- "8000:8000"
depends_on:
- redis
And a minimal Dockerfile:
FROM python:3.12-slim
WORKDIR /code
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
Now run:
docker compose up --build
The API will be reachable at http://localhost:8000. You can test the WebSocket endpoint with a tool like websocat or a simple HTML client.
Building a Quick HTML Client
For a hands‑on demo, create client.html in the project root. It connects to the WebSocket, displays incoming messages, and lets you send new ones.
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>FastAPI Chat Demo</title>
<style>
body { font-family: Arial, sans-serif; margin: 2rem; }
#log { border: 1px solid #ccc; height: 300px; overflow-y: scroll; padding: .5rem; }
#msg { width: 80%; }
</style>
</head>
<body>
<h2>Room: general</h2>
<div id="log"></div>
<input id="msg" placeholder="Type a message..." autocomplete="off"/>
<button id="send">Send</button>
<script>
const room = "general";
const ws = new WebSocket(`ws://${location.host}/ws/chat/${room}`);
const log = document.getElementById('log');
const input = document.getElementById('msg');
const btn = document.getElementById('send');
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
const line = document.createElement('div');
line.textContent = `[${data.timestamp}] ${data.client_id.slice(0,8)}: ${data.message}`;
log.appendChild(line);
log.scrollTop = log.scrollHeight;
};
btn.onclick = () => {
if (input.value.trim()) {
ws.send(input.value);
input.value = '';
}
};
</script>
</body>
</html>
Open client.html in two browser tabs. Type a message in one tab; it instantly appears in the other, confirming that the WebSocket‑Redis pipeline works end‑to‑end.
Note: In production, always serve the static HTML via a CDN or a dedicated static file server. Exposing the WebSocket endpoint behind a reverse proxy (e.g., Nginx) also adds TLS termination and rate‑limiting.
Scaling the Service
When you need to handle thousands of concurrent connections, consider these strategies:
- Horizontal FastAPI workers: Deploy multiple instances behind a load balancer. Because each worker subscribes to the same Redis channel, messages propagate to all clients regardless of which worker accepted the connection.
- Redis clustering: Split the data across multiple nodes to avoid a single point of failure and to increase throughput.
- Connection limits: Tune
uvicornworkers andmax_concurrent_connectionsin your ASGI server to match your hardware. - Back‑pressure handling: Implement a queue (e.g., Redis Streams) if you anticipate bursts that could overwhelm the subscriber loop.
These patterns are common in large‑scale chat platforms like Slack or Discord, and they translate well to any real‑time notification system.
Persisting Chat History
Our demo currently discards messages after broadcast. To add persistence, you can push each validated ChatMessage into a Redis List or a dedicated database (PostgreSQL, MongoDB, etc.). Here’s a lightweight approach using a Redis List per room:
# Append to history after validation
await redis.rpush(f"history:{room_id}", json.dumps(payload.dict()))
# Trim to last 100 messages
await redis.ltrim(f"history:{room_id}", -100, -1)
When a new client connects, you can preload the recent history:
history = await redis.lrange(f"history:{room_id}", 0, -1)
for entry in history:
await websocket.send_text(entry)
This gives users context without requiring a full‑blown database. For compliance or analytics, however, a relational store is advisable.
Testing the WebSocket Layer
Automated testing of async WebSockets can be done with httpx.AsyncClient and pytest-asyncio. Below is a minimal test suite that verifies message round‑trip and validation.
import pytest
import asyncio
from httpx import AsyncClient
from fastapi import WebSocket
from app.main import app
@pytest.mark.asyncio
async def test_chat_flow():
async with AsyncClient(app=app, base_url="http://test") as client:
async with client.websocket_connect("/ws/chat/testroom") as ws:
# Send a valid message
await ws.send_text("Hello, world!")
response = await ws.receive_text()
data = json.loads(response)
assert data["message"] == "Hello, world!"
assert data["room_id"] == "testroom"
# Send an invalid (empty) message
await ws.send_text("")
error = await ws.receive_text()
err_data = json.loads(error)
assert "error" in err_data
Running pytest -q will spin up the ASGI app in memory, ensuring your WebSocket logic stays reliable as you iterate.
Deploying to the Cloud
For production, containerize the app (as we already did) and push the image to a registry (Docker Hub, GitHub Packages, etc.). Then deploy using a managed service like AWS Fargate, Google Cloud Run, or Azure Container Apps. All three support WebSocket traffic out of the box when you configure the appropriate ingress.
Example snippet for a Cloud Run service:
gcloud run deploy fastapi-chat \
--image gcr.io/PROJECT_ID/fastapi-chat:latest \
--platform managed \
--port 8000 \
--allow-unauthenticated \
--region us-central1
Don’t forget to provision a Redis instance (e.g., AWS ElastiCache, Google Memorystore) and set the REDIS_HOST environment variable accordingly. Your container should read the host from os.getenv instead of hard‑coding “redis”.
Security Considerations
Real‑time services are attractive targets, so implement these safeguards:
- Authentication: Use JWT tokens passed as a query param or header, and verify them before accepting the WebSocket.
- Rate limiting: Apply per‑IP or per‑user limits using a middleware like
slowapi. - Origin checking: Reject connections from unknown origins to mitigate CSRF‑style attacks.
- Message sanitization: Escape HTML or use a library like
bleachto prevent XSS when rendering messages on the client.
These measures keep your chat service robust against both accidental misuse and malicious actors.
Monitoring & Observability
Instrument the application with Prometheus metrics and OpenTelemetry traces. FastAPI integrates nicely with starlette_exporter for HTTP metrics, while you can emit custom counters for WebSocket connections:
from prometheus_client import Counter
ws_connections = Counter("ws_active_connections", "Current active WebSocket connections")
ws_messages = Counter("ws_messages_total", "Total messages processed")
# Increment on connect
ws_connections.inc()
# Decrement on disconnect
ws_connections.dec()
# Increment on each broadcast
ws_messages.inc()
Collecting these metrics helps you spot spikes, memory leaks, or network bottlenecks before they impact users.