Skip to main content

FastAPI WebSocket Example

Introduction​

WebSockets provide a persistent connection between a client and server, allowing for real-time, bidirectional communication. Unlike traditional HTTP requests where the client must initiate all interactions, WebSockets enable both the client and server to send messages at any time once a connection is established.

In this tutorial, we'll build a real-time chat application using FastAPI's WebSocket support. This example will demonstrate:

  1. Setting up WebSocket endpoints in FastAPI
  2. Managing WebSocket connections
  3. Broadcasting messages to connected clients
  4. Creating a simple HTML/JavaScript frontend to interact with our WebSocket server

By the end of this tutorial, you'll have a functional real-time chat application that you can adapt for your own projects.

Prerequisites​

Before starting, ensure you have:

  • Python 3.7+ installed
  • Basic understanding of FastAPI
  • Basic knowledge of HTML and JavaScript

Let's begin by installing the required packages:

bash
pip install fastapi uvicorn

Building a Simple Chat Application​

Step 1: Create the FastAPI Application​

First, let's create a basic FastAPI application with WebSocket support. Create a file named app.py:

python
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from typing import List

app = FastAPI()

# HTML content for our chat interface
html = """
<!DOCTYPE html>
<html>
    <head>
        <title>FastAPI Chat</title>
        <style>
            body {
                font-family: Arial, sans-serif;
                max-width: 800px;
                margin: 0 auto;
                padding: 1em;
            }
            #messages {
                border: 1px solid #ddd;
                height: 400px;
                overflow-y: auto;
                padding: 10px;
                margin-bottom: 10px;
                border-radius: 4px;
            }
            #messageForm {
                display: flex;
            }
            #messageInput {
                flex: 1;
                padding: 8px;
                border: 1px solid #ddd;
                border-radius: 4px;
            }
            button {
                background-color: #4CAF50;
                color: white;
                border: none;
                padding: 9px 15px;
                margin-left: 10px;
                border-radius: 4px;
                cursor: pointer;
            }
            .message {
                margin-bottom: 5px;
                padding: 5px;
            }
            .user-message {
                background-color: #e9f5ff;
                border-radius: 4px;
            }
            .system-message {
                color: #666;
                font-style: italic;
            }
        </style>
    </head>
    <body>
        <h1>FastAPI Chat Example</h1>
        <div id="messages"></div>
        <form id="messageForm">
            <input type="text" id="messageInput" placeholder="Type your message here" autocomplete="off"/>
            <input type="text" id="usernameInput" placeholder="Your username" autocomplete="off"/>
            <button type="submit">Send</button>
        </form>
        
        <script>
            const messagesDiv = document.getElementById('messages');
            const messageForm = document.getElementById('messageForm');
            const messageInput = document.getElementById('messageInput');
            const usernameInput = document.getElementById('usernameInput');
            
            // Create WebSocket connection
            const ws = new WebSocket(`ws://${window.location.host}/ws`);
            
            ws.onmessage = function(event) {
                const message = event.data;
                const messageElement = document.createElement('div');
                
                if (message.startsWith('User')) {
                    messageElement.className = 'message user-message';
                } else {
                    messageElement.className = 'message system-message';
                }
                
                messageElement.textContent = message;
                messagesDiv.appendChild(messageElement);
                messagesDiv.scrollTop = messagesDiv.scrollHeight;
            };
            
            messageForm.addEventListener('submit', function(event) {
                event.preventDefault();
                
                const message = messageInput.value;
                const username = usernameInput.value || 'Anonymous';
                
                if (message) {
                    ws.send(JSON.stringify({
                        username: username,
                        message: message
                    }));
                    messageInput.value = '';
                }
            });
        </script>
    </body>
</html>
"""

# Class to manage WebSocket 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 send_personal_message(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.get("/")
async def get():
    return HTMLResponse(html)


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        # Send a welcome message to the new client
        await manager.send_personal_message("Connected to the chat server!", websocket)
        # Broadcast that a new user has joined
        await manager.broadcast(f"User #{len(manager.active_connections)} has joined the chat")
        
        while True:
            data = await websocket.receive_json()
            username = data.get("username", "Anonymous")
            message = data.get("message", "")
            
            # Broadcast the message to all connected clients
            await manager.broadcast(f"User {username}: {message}")
            
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"User #{len(manager.active_connections) + 1} has left the chat")

Step 2: Run the Application​

Now, let's run our FastAPI application:

bash
uvicorn app:app --reload

Visit http://localhost:8000 in your web browser to see the chat interface. Open multiple browser windows to simulate different users chatting with each other.

Understanding the Code​

Let's break down what's happening in our chat application:

The HTML Interface​

The HTML content includes:

  1. A styled chat interface with a message display area
  2. A form for sending messages
  3. Input fields for the message and username
  4. JavaScript code to:
    • Establish the WebSocket connection
    • Handle incoming messages
    • Send messages when the form is submitted

The ConnectionManager Class​

This class is crucial for managing WebSocket connections:

  • active_connections stores all connected clients
  • connect() accepts a new WebSocket connection and adds it to the list
  • disconnect() removes a connection from the list
  • send_personal_message() sends a message to a specific client
  • broadcast() sends a message to all connected clients

The WebSocket Endpoint​

The /ws endpoint handles WebSocket connections:

  1. When a new client connects:

    • The connection is accepted via manager.connect()
    • A welcome message is sent to the client
    • A message is broadcast to all clients about the new user

  2. While the connection is active:

    • The endpoint receives JSON data containing a username and message
    • The message is broadcast to all connected clients

  3. When a client disconnects:

    • The connection is removed via manager.disconnect()
    • A message is broadcast about the user leaving

Enhancing the Example​

Let's extend our example by adding a few more features:

Step 3: Add User Typing Indicators​

Update app.py to add typing indicators:

python
# Add this to the existing WebSocket endpoint
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        await manager.send_personal_message("Connected to the chat server!", websocket)
        await manager.broadcast(f"User #{len(manager.active_connections)} has joined the chat")
        
        while True:
            data = await websocket.receive_json()
            username = data.get("username", "Anonymous")
            message = data.get("message", "")
            typing = data.get("typing", False)
            
            if typing:
                # Only send typing indicators to other users
                for connection in manager.active_connections:
                    if connection != websocket:
                        await connection.send_text(f"{username} is typing...")
            elif message:
                # Broadcast the message to all connected clients
                await manager.broadcast(f"User {username}: {message}")
            
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"User #{len(manager.active_connections) + 1} has left the chat")

And update the JavaScript in the HTML content:

javascript
// Add this after the existing JavaScript
let typingTimer;

messageInput.addEventListener('input', function() {
    clearTimeout(typingTimer);
    
    // Send typing indicator
    ws.send(JSON.stringify({
        username: usernameInput.value || 'Anonymous',
        typing: true
    }));
    
    // Stop sending typing indicator after 1 second of inactivity
    typingTimer = setTimeout(() => {
        ws.send(JSON.stringify({
            username: usernameInput.value || 'Anonymous',
            typing: false
        }));
    }, 1000);
});

Real-World Applications​

WebSockets in FastAPI can be used for various real-time applications:

  1. Collaborative Editing Tools: Multiple users can edit documents simultaneously with changes reflected in real time.

  2. Live Dashboards: Update metrics and charts without requiring page refreshes.

  3. Multiplayer Games: Synchronize game states across multiple players.

  4. Notification Systems: Push notifications to users instantly when events occur.

  5. IoT Applications: Monitor and control IoT devices with minimal latency.

Performance Considerations​

When working with WebSockets in production, keep the following in mind:

  1. Connection Limits: Each WebSocket connection consumes server resources, so implement a strategy for limiting connections if needed.

  2. Authentication: For secure applications, authenticate users before allowing WebSocket connections.

  3. Heartbeats: Implement periodic heartbeat messages to detect and clean up stale connections.

  4. Rate Limiting: Prevent abuse by implementing rate limits for message sending.

  5. Scaling: Consider using Redis or other message brokers to broadcast messages across multiple server instances.

Summary​

In this tutorial, we've built a real-time chat application using FastAPI's WebSocket support. We've learned how to:

  1. Create WebSocket endpoints in FastAPI
  2. Manage multiple WebSocket connections
  3. Send and broadcast messages
  4. Create a simple frontend to interact with our WebSocket server
  5. Add enhanced features like typing indicators

WebSockets provide a powerful way to implement real-time features in your web applications. FastAPI makes it straightforward to add WebSocket functionality with its simple and intuitive API.

Additional Exercises​

  1. Add message persistence using a database (SQLite, PostgreSQL, etc.)
  2. Implement private messaging between specific users
  3. Add user authentication before allowing WebSocket connections
  4. Create separate chat rooms or channels
  5. Add file sharing capabilities
  6. Implement message read receipts

Additional Resources​

With these concepts and examples, you should now be well-equipped to implement WebSockets in your own FastAPI applications!


If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)