WebSockets

Learn essential concepts of websockets in FastAPI.

What You'll Learn

This section covers WebSocket fundamentals, including:

• WebSocket connections vs HTTP requests

• Real-time bidirectional communication

• Connection lifecycle management

• Broadcasting messages to multiple clients

• Error handling and disconnection

• Chat applications and live updates

Core Concepts

WebSockets enable persistent, bidirectional communication between client and server. Unlike HTTP's request-response model, WebSockets maintain open connections for real-time data exchange.

Key Ideas

Understanding WebSockets helps you:

• Build real-time applications (chat, notifications, dashboards)

• Reduce latency compared to polling

• Handle persistent connections efficiently

• Broadcast updates to multiple connected clients

• Implement live collaboration features

• Create responsive user experiences

Basic WebSocket Connection

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            # Receive message from client
            data = await websocket.receive_text()
            # Send response back
            await websocket.send_text(f"Echo: {data}")
    except Exception:
        await websocket.close()

Chat Application Example

from fastapi import WebSocket

# Store active connections
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):
        # Send message to all connected clients
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat/{user_id}")
async def websocket_chat(websocket: WebSocket, user_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"User {user_id}: {data}")
    except Exception:
        manager.disconnect(websocket)

Live Notification Example

@app.websocket("/ws/notifications/{client_id}")
async def websocket_notifications(websocket: WebSocket, client_id: str):
    await websocket.accept()
    try:
        while True:
            # Receive acknowledgment or keep connection alive
            await websocket.receive_text()
    except Exception:
        await websocket.close()

@app.post("/notify/{client_id}")
async def send_notification(client_id: str, message: str):
    # Send notification to specific client
    # (requires connection tracking in production)
    return {"status": "notification_sent", "client": client_id}

Real-World Usage

WebSockets enable:

• **Chat Applications**: Real-time messaging between users

• **Live Notifications**: Instant updates without polling

• **Collaborative Tools**: Shared document editing, whiteboards

• **Live Dashboards**: Real-time data visualization and monitoring

• **Gaming**: Multiplayer interactions and state synchronization

• **Notifications**: Push alerts and real-time status updates

🔑 Key Takeaways

• ✅ Understand the purpose of websockets

• ✅ Know when to apply this pattern

• ✅ Follow best practices consistently

• ✅ Test thoroughly in production scenarios

Next step: Explore the advanced section for production patterns and optimization techniques.