TOP 5 March 2, 2026, 11:30 a.m.

Tech Tutorial - March 02 2026 113008

Welcome back, Codeyaan explorers! Today we’re diving into a hands‑on tutorial that takes you from zero to a fully functional real‑time chat application using FastAPI and WebSockets. By the end of this guide you’ll have a production‑ready backend, a sleek JavaScript front‑end, and a handful of pro tips that will make your next project faster, safer, and more scalable.

Why FastAPI and WebSockets?

FastAPI has become the go‑to framework for modern Python APIs thanks to its speed, type safety, and automatic OpenAPI docs. Pair it with WebSockets—a protocol that enables bidirectional, low‑latency communication—and you get the perfect combo for live chat, gaming, or collaborative tools. Unlike classic HTTP polling, WebSockets keep a single TCP connection open, dramatically reducing overhead and latency.

In real‑world scenarios, think of customer support dashboards, live sports scoreboards, or collaborative code editors. All of these demand instant feedback, and FastAPI’s async support makes handling thousands of concurrent sockets a breeze.

Setting Up Your Development Environment

Before we write any code, let’s ensure your workstation is ready. The tutorial assumes you have Python 3.11+ installed. If you’re on Windows, macOS, or Linux, the steps are identical.

Create a virtual environment: python -m venv venv
Activate it:
- Windows: venv\Scripts\activate
- macOS/Linux: source venv/bin/activate
Install dependencies: pip install fastapi[all] uvicorn python‑dotenv

We’ll also use python‑dotenv to keep secrets like JWT keys out of source control. Create a .env file in the project root with the following content:

# .env
SECRET_KEY=supersecretkey123
HOST=127.0.0.1
PORT=8000

Creating the FastAPI Server

Let’s scaffold the basic FastAPI app. This file will host our HTTP routes, the WebSocket endpoint, and a simple authentication layer.

# main.py
import os
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from dotenv import load_dotenv
import jwt
import asyncio

load_dotenv()
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

SECRET_KEY = os.getenv("SECRET_KEY")
HOST = os.getenv("HOST", "127.0.0.1")
PORT = int(os.getenv("PORT", 8000))

# In‑memory store for active connections
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()

Notice the use of type hints and async methods—FastAPI will automatically generate OpenAPI docs for the HTTP endpoints we add later.

Simple JWT Authentication

While WebSockets can work without auth, a production chat app needs to know who’s talking. We’ll issue a JWT token via a classic OAuth2 password flow, then validate it on each WebSocket connection.

# Authentication utilities
def create_access_token(username: str) -> str:
    payload = {"sub": username}
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def decode_token(token: str) -> str:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload.get("sub")
    except jwt.PyJWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
                            detail="Invalid token")

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # In a real app, verify username/password against a DB
    if form_data.username == "demo" and form_data.password == "demo":
        access_token = create_access_token(form_data.username)
        return {"access_token": access_token, "token_type": "bearer"}
    raise HTTPException(status_code=400, detail="Incorrect credentials")

Now we have a token endpoint that the front‑end can call to retrieve a JWT. The secret key lives in .env, keeping it out of the repo.

Implementing the WebSocket Endpoint

The heart of our chat lives here. Each client connects, sends messages, and receives broadcasts from all other participants.

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(oauth2_scheme)):
    # Authenticate the user
    username = decode_token(token)
    await manager.connect(websocket)
    await manager.broadcast(f"🔔 {username} joined the chat")
    try:
        while True:
            data = await websocket.receive_text()
            message = f"{username}: {data}"
            await manager.broadcast(message)
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"⚪ {username} left the chat")

Key points to note:

Dependency injection: FastAPI automatically extracts the token from the query string or Authorization header.
Async loop: The while True loop listens for incoming text messages without blocking other connections.
Graceful disconnect: On WebSocketDisconnect we clean up and inform remaining users.

Running the Server

Launch the app with Uvicorn, which is built for async workloads:

# In the terminal
uvicorn main:app --host $HOST --port $PORT --reload

The --reload flag watches for file changes, perfect for development.

Building the Front‑End

Our front‑end will be a single HTML file powered by vanilla JavaScript. No heavy frameworks required, keeping the example lightweight and easy to understand.

<!-- index.html -->
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>FastAPI Chat Demo</title>
    <style>
        body {font-family: Arial, sans-serif; margin: 2rem;}
        #chat {border: 1px solid #ccc; height: 300px; overflow-y: scroll; padding: 0.5rem;}
        #msg {width: 80%;}
    </style>
</head>
<body>
    <h2>FastAPI Real‑Time Chat</h2>
    <div id="login">
        <input id="username" placeholder="Username">
        <input id="password" type="password" placeholder="Password">
        <button id="loginBtn">Login</button>
    </div>
    <div id="chatSection" style="display:none;">
        <div id="chat"></div>
        <input id="msg" placeholder="Type a message...">
        <button id="sendBtn">Send</button>
    </div>
    <script>
        const loginBtn = document.getElementById('loginBtn');
        const sendBtn = document.getElementById('sendBtn');
        const chatDiv = document.getElementById('chat');
        let ws;
        let token = '';

        loginBtn.onclick = async () => {
            const username = document.getElementById('username').value;
            const password = document.getElementById('password').value;
            const resp = await fetch('/token', {
                method: 'POST',
                headers: {'Content-Type': 'application/x-www-form-urlencoded'},
                body: new URLSearchParams({username, password})
            });
            const data = await resp.json();
            token = data.access_token;
            initWebSocket();
            document.getElementById('login').style.display = 'none';
            document.getElementById('chatSection').style.display = 'block';
        };

        function initWebSocket() {
            ws = new WebSocket(`ws://${window.location.host}/ws/chat?token=${token}`);
            ws.onmessage = (event) => {
                const p = document.createElement('p');
                p.textContent = event.data;
                chatDiv.appendChild(p);
                chatDiv.scrollTop = chatDiv.scrollHeight;
            };
            ws.onclose = () => alert('Connection closed');
        }

        sendBtn.onclick = () => {
            const msg = document.getElementById('msg').value;
            ws.send(msg);
            document.getElementById('msg').value = '';
        };
    </script>
</body>
</html>

The script first logs in via the /token endpoint, stores the JWT, and then opens a WebSocket connection that includes the token as a query parameter. Each incoming message is appended to the chat window, and the scroll is auto‑adjusted.

Serving Static Files

FastAPI can serve static assets directly. Add this snippet to main.py to expose index.html and any CSS/JS files placed in a static folder.

from fastapi.staticfiles import StaticFiles

app.mount("/", StaticFiles(directory="static", html=True), name="static")

Now navigating to http://127.0.0.1:8000/ will load the chat UI automatically.

Testing Your Chat Application

Manual testing is straightforward: open two browser tabs, log in with the same credentials (or create a quick “user switch” UI), and watch messages appear instantly. For automated testing, FastAPI’s TestClient can simulate HTTP token retrieval, while websockets library can validate the socket flow.

# test_chat.py
import asyncio
import pytest
from httpx import AsyncClient
from websockets import connect
from main import app

@pytest.mark.asyncio
async def test_full_flow():
    async with AsyncClient(app=app, base_url="http://test") as client:
        # 1️⃣ Get JWT
        resp = await client.post("/token", data={"username":"demo","password":"demo"})
        token = resp.json()["access_token"]

        # 2️⃣ Open two WebSocket connections
        ws_url = f"ws://test/ws/chat?token={token}"
        async with connect(ws_url) as ws1, connect(ws_url) as ws2:
            await ws1.send("Hello from ws1")
            msg = await ws2.recv()
            assert "demo: Hello from ws1" in msg

This test confirms that a message sent from one socket is broadcast to another, proving our manager works as intended.

Scaling Considerations

While the in‑memory ConnectionManager works for demos, production deployments often require a distributed approach. Here are three common strategies:

Redis Pub/Sub: Publish each message to a Redis channel; all server instances subscribe and broadcast locally. This decouples sockets across multiple workers.
Message Queues (Kafka/RabbitMQ): Useful when you need durability, replay, or complex routing logic.
Server‑Sent Events (SSE) fallback: For environments where WebSocket support is limited, an SSE fallback can keep the UI functional.

Implementing Redis is the most common path for FastAPI. Below is a minimal snippet that replaces the in‑memory list with a Redis channel.

# redis_manager.py
import aioredis
import json

class RedisConnectionManager:
    def __init__(self, redis_url="redis://localhost"):
        self.redis = aioredis.from_url(redis_url)
        self.channel = "chat"

    async def broadcast(self, message: str):
        await self.redis.publish(self.channel, json.dumps({"msg": message}))

    async def subscriber(self):
        pubsub = self.redis.pubsub()
        await pubsub.subscribe(self.channel)
        async for msg in pubsub.listen():
            if msg["type"] == "message":
                yield json.loads(msg["data"])["msg"]

Integrating this manager involves spawning a background task that forwards Redis messages to each connected WebSocket. The pattern stays the same; only the storage layer changes.

Pro Tip: When scaling with Docker or Kubernetes, expose the WebSocket port via an Ingress that supports upgrade headers. Nginx Ingress Controller with proxy_http_version 1.1 and proxy_set_header Upgrade $http_upgrade is a battle‑tested combo.

Security Hardening

Security is non‑negotiable for any public‑facing chat service. Below are three quick hardening steps you can add without rewriting the core logic.

Origin Check: Verify the Origin header inside the WebSocket handshake to prevent cross‑site socket hijacking.
Rate Limiting: Use slowapi or redis‑rate‑limit to throttle message bursts per user.
Message Sanitization: Strip HTML tags or use a library like bleach to avoid XSS attacks when rendering messages in the browser.

Here’s a concise example of an origin guard inside the endpoint:

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(oauth2_scheme)):
    if websocket.headers.get("origin") != "http://localhost:8000":
        await websocket.close(code=1008)  # Policy Violation
        return
    # …rest of the logic remains unchanged

Deploying to Production

For a production‑grade deployment, we recommend using Gunicorn with the uvicorn.workers.UvicornWorker class. This setup gives you multiple worker processes, each capable of handling async tasks.

# Install Gunicorn
pip install gunicorn

# Run 4 workers on port 80
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:80

Combine this with a reverse proxy (NGINX or Traefik) that terminates TLS, handles static file caching, and forwards WebSocket upgrades. A typical NGINX block looks like:

server {
    listen 443 ssl;
    server_name chat.example.com;

    ssl_certificate /etc/letsencrypt/live/chat.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/chat.example.com/privkey.pem;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;



    
    
        
            
                
            
        
    

    
    
        
            Share this article