Python WebSockets — Real-Time Communication

WebSockets enable bidirectional, real-time communication between client and server. Unlike HTTP (request-response), WebSockets maintain a persistent connection for live updates.

Learning Objectives

• Understand WebSocket vs HTTP
• Build WebSocket servers with FastAPI
• Handle concurrent connections with a connection manager
• Implement chat and notification systems

WebSocket vs HTTP

HTTP (Request-Response):
Client ──► "Give me data" ──► Server
Client ◄── "Here is data" ◄── Server
(Connection closes)

WebSocket (Persistent):
Client ◄──── Persistent Connection ────► Server
Client ──► "Hello" ──────────────────────► Server
Client ◄── "Response" ◄────────────────── Server
Client ◄── "Update"  ◄────────────────── Server (server can push!)
Client ──► "Message" ────────────────────► Server

Use cases: Chat apps, live dashboards, multiplayer games, real-time notifications, live sports scores.

FastAPI WebSocket Server

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    """Manages active WebSocket connections."""

    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 send_personal(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

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

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket)
    try:
        # Notify everyone
        await manager.broadcast(f"{client_id} joined the chat")

        while True:
            # Receive message from client
            data = await websocket.receive_text()

            # Echo back to sender
            await manager.send_personal(f"You said: {data}", websocket)

            # Broadcast to all others
            await manager.broadcast(f"{client_id}: {data}")

    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"{client_id} left the chat")

WebSocket Client

import asyncio
import websockets

async def client():
    async with websockets.connect("ws://localhost:8000/ws/alice") as ws:
        # Send message
        await ws.send("Hello, server!")

        # Receive response
        response = await ws.recv()
        print(f"Received: {response}")

asyncio.run(client())

Broadcasting Notifications

class NotificationManager:
    def __init__(self):
        self.subscriptions = {}  # user_id -> websocket

    async def subscribe(self, user_id: str, ws: WebSocket):
        await ws.accept()
        self.subscriptions[user_id] = ws

    async def notify(self, user_id: str, message: str):
        if user_id in self.subscriptions:
            await self.subscriptions[user_id].send_text(message)

    async def broadcast(self, message: str):
        for ws in self.subscriptions.values():
            await ws.send_text(message)

Key Takeaways

1. WebSockets are full-duplex (both directions simultaneously)
2. Use WebSockets for real-time features (chat, gaming, live data)
3. Handle disconnections gracefully with try/except
4. Use connection managers to track active connections
5. Broadcast messages to all connected clients when needed
6. Consider authentication for WebSocket connections
7. Use ping/pong to detect stale connections