codeYaan Logo Blog
Home Blog Services Courses
Login
Tech Tutorial - February 28 2026 233007
AI TOOLS Feb. 28, 2026, 11:30 p.m.

Tech Tutorial - February 28 2026 233007

Welcome to today’s deep‑dive tutorial! In the next few minutes we’ll walk through building a production‑ready real‑time chat application using FastAPI, WebSockets, and Redis. By the end of this guide you’ll have a fully functional backend, a lightweight JavaScript client, and a handful of best‑practice tricks you can reuse in any real‑time project.

Why FastAPI and WebSockets?

FastAPI has become the go‑to framework for high‑performance APIs thanks to its async‑first design and automatic OpenAPI documentation. Pairing it with native WebSocket support lets you push data to browsers without the overhead of polling. This combination delivers sub‑millisecond latency, which is essential for chat, live dashboards, and collaborative tools.

Redis enters the picture as a fast, in‑memory message broker. Using its Pub/Sub feature we can decouple the WebSocket connections from the core business logic, making the system horizontally scalable across multiple worker processes or containers.

Setting Up the Development Environment

Before we write any code, let’s get the tooling ready. The stack includes Python 3.12, FastAPI, Uvicorn, and Redis. If you’re on macOS or Linux, the commands below should work out of the box; Windows users can use WSL or Docker.

  • Install Python 3.12 (or newer) from python.org.
  • Create a virtual environment:
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
  • Install the required packages:
pip install fastapi[all] uvicorn redis asyncio

Finally, spin up a local Redis instance. The simplest way is via Docker:

docker run -d --name redis -p 6379:6379 redis:7-alpine

With the environment ready, we can start coding the server.

Building the FastAPI Server

The server will expose two endpoints: a standard HTTP health‑check and a WebSocket route for chat communication. Let’s create a new file called main.py and add the basic scaffolding.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import asyncio
import json
import redis.asyncio as aioredis

app = FastAPI()
redis = aioredis.from_url("redis://localhost:6379", decode_responses=True)

@app.get("/", response_class=HTMLResponse)
async def get():
    return """
    <!DOCTYPE html>
    <html>
    <head><title>FastAPI Chat</title></head>
    <body>
        <h2>Real‑time Chat</h2>
        <div id="messages"></div>
        <input id="msgInput" type="text" placeholder="Type a message..." autofocus/>
        <script src="/static/chat.js"></script>
    </body>
    </html>
    """

The HTML snippet above is deliberately minimal; it loads a JavaScript file we’ll create later. The next step is to implement the WebSocket endpoint that will interact with Redis.

Managing Connections

We need a lightweight connection manager that tracks active sockets and broadcasts incoming messages to all participants. The manager also subscribes to a Redis channel so that messages from any server instance get relayed.

class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

WebSocket Route with Redis Pub/Sub

Now we tie everything together. Each incoming WebSocket connection will subscribe to a Redis channel called chat. When a client sends a message, we publish it to Redis; all subscribers—including the sender—receive the broadcast.

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    pubsub = redis.pubsub()
    await pubsub.subscribe("chat")

    async def redis_listener():
        async for message in pubsub.listen():
            if message["type"] == "message":
                await manager.broadcast(message["data"])

    listener_task = asyncio.create_task(redis_listener())

    try:
        while True:
            data = await websocket.receive_text()
            payload = json.dumps({"msg": data})
            await redis.publish("chat", payload)
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await pubsub.unsubscribe("chat")
        listener_task.cancel()
    finally:
        await pubsub.close()

Notice the use of asyncio.create_task to run the Redis listener in the background. This pattern ensures we never block the main event loop, keeping latency low even under heavy load.

Creating the Front‑End Client

Our front‑end is a single JavaScript file that opens a WebSocket connection, renders incoming messages, and sends user input back to the server. Place this file in a static folder so FastAPI can serve it automatically.

const ws = new WebSocket(`ws://${location.host}/ws/chat`);
const messagesDiv = document.getElementById('messages');
const input = document.getElementById('msgInput');

ws.onmessage = (event) => {
    const data = JSON.parse(event.data);
    const p = document.createElement('p');
    p.textContent = data.msg;
    messagesDiv.appendChild(p);
    messagesDiv.scrollTop = messagesDiv.scrollHeight;
};

input.addEventListener('keydown', (e) => {
    if (e.key === 'Enter' && input.value.trim() !== '') {
        ws.send(input.value.trim());
        input.value = '';
    }
});

The script is intentionally terse: it handles JSON parsing, auto‑scrolls the message pane, and sends messages on the Enter key. You can extend it with usernames, emojis, or typing indicators without changing the server logic.

Running the Application

Start the FastAPI server with Uvicorn, enabling reload for development convenience:

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Open http://localhost:8000 in two separate browser tabs. Type a message in one tab, and watch it appear instantly in the other—thanks to the WebSocket + Redis pipeline we just built.

Pro Tip: When deploying to production, replace the in‑process ConnectionManager with a distributed store (e.g., Redis Streams or a dedicated message broker) so that each instance can discover every active socket across the cluster.

Scaling Out with Docker Compose

In real‑world scenarios you’ll rarely run a single process. Let’s see how to orchestrate multiple FastAPI workers and a shared Redis instance using Docker Compose. Create a docker-compose.yml file with the following services:

version: "3.9"
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
  api:
    build: .
    command: uvicorn main:app --host 0.0.0.0 --port 8000
    volumes:
      - .:/app
    ports:
      - "8000:8000"
    depends_on:
      - redis
    environment:
      - REDIS_URL=redis://redis:6379

Update main.py to read the Redis URL from the environment variable, which makes the container‑based setup seamless.

import os
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
redis = aioredis.from_url(REDIS_URL, decode_responses=True)

Run docker-compose up --build and you’ll have a load‑balanced chat service ready to accept connections from any number of containers. Because each instance subscribes to the same Redis channel, messages flow across all workers automatically.

Note: For high traffic you might want to enable Redis clustering or switch to a more robust broker like NATS or RabbitMQ, especially if you need message persistence or guaranteed delivery.

Real‑World Use Cases

While a simple chat app is a great learning project, the pattern we built scales to many production scenarios:

  1. Collaborative Document Editing: Broadcast cursor positions, text diffs, and version stamps via the same Pub/Sub channel.
  2. Live Sports Scoreboards: Push score updates from a backend data feed to thousands of viewers without page reloads.
  3. IoT Dashboards: Stream sensor readings from edge devices to a central UI in near‑real‑time.

In each case the core idea stays the same: a lightweight FastAPI layer handles WebSocket connections, while Redis (or a comparable broker) distributes messages across a horizontally scalable fleet.

Security Considerations

Real‑time endpoints are attractive targets for abuse. Here are three quick safeguards you should implement before going public:

  • Authentication Tokens: Require a JWT in the WebSocket query string and validate it on connection.
  • Rate Limiting: Use a token bucket algorithm stored in Redis to throttle message bursts per user.
  • Origin Checks: Reject connections from unknown origins to mitigate CSRF‑style attacks.

Below is a concise example of JWT validation inside the WebSocket route.

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

def get_current_user(token: HTTPAuthorizationCredentials = Depends(security)):
    try:
        payload = jwt.decode(token.credentials, "YOUR_SECRET_KEY", algorithms=["HS256"])
        return payload["sub"]
    except jwt.PyJWTError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid authentication credentials"
        )

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket, user: str = Depends(get_current_user)):
    # user variable now holds the authenticated username
    await manager.connect(websocket)
    # ... rest of the logic remains unchanged ...
Pro Tip: Store the secret key in an environment variable or a secret manager (AWS Secrets Manager, HashiCorp Vault) rather than hard‑coding it.

Testing the Real‑Time Flow

Automated testing of WebSocket interactions can be tricky, but FastAPI’s TestClient combined with anyio makes it doable. Below is a minimal test that ensures a message published by one client is received by another.

import pytest
from fastapi.testclient import TestClient
from main import app, redis

client = TestClient(app)

@pytest.mark.asyncio
async def test_chat_broadcast():
    async with client.websocket_connect("/ws/chat") as ws1:
        async with client.websocket_connect("/ws/chat") as ws2:
            await ws1.send_text("Hello from ws1")
            data = await ws2.receive_text()
            assert "Hello from ws1" in data

    # Clean up Redis channel
    await redis.flushdb()

This test spins up two in‑process WebSocket connections, sends a message from the first, and verifies the second receives it. Running it as part of your CI pipeline gives you confidence that the core real‑time pipeline stays intact after future changes.

Performance Tweaks

When you start handling thousands of concurrent sockets, a few low‑level adjustments can make a big difference:

  • Use Uvicorn with --workers: Spawns multiple processes to utilize all CPU cores.
  • Enable TCP Fast Open: Reduces handshake latency for new connections.
  • Set max_message_size in WebSocket: Prevents oversized payloads from exhausting memory.

Example launch command with 4 workers:

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 --ws-max-size 1048576

Conclusion

We’ve covered everything from environment setup to production‑grade scaling for a real‑time chat service built with FastAPI, WebSockets, and Redis. By leveraging async I/O, a simple Pub/Sub broker, and a clean connection manager, you now have a solid foundation for any low‑latency, high‑throughput application. Remember to secure your endpoints, monitor performance, and keep your Redis deployment resilient. Happy coding, and may your messages always arrive instantly!

Share this article
Back to Blog