Broadcasting to multiple clients in FastAPI

Introduction

Broadcasting lets a server send the same message to many clients at once. This helps keep everyone updated together.

You want to send live chat messages to all users in a chat room.

You need to update multiple users with real-time notifications.

You want to broadcast live scores or updates in a game app.

You want to push stock price changes to many clients simultaneously.

Syntax

FastAPI

from fastapi import FastAPI, WebSocket
from fastapi.websockets import WebSocketDisconnect

app = FastAPI()

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)

The ConnectionManager class keeps track of all connected clients.

The broadcast method sends a message to every connected client.

Examples

This example accepts WebSocket connections, receives messages, and broadcasts them to all clients.

FastAPI

manager = ConnectionManager()

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f'Message: {data}')
    except WebSocketDisconnect:
        manager.disconnect(websocket)

This line sends the text 'Hello everyone!' to all connected clients.

FastAPI

await manager.broadcast('Hello everyone!')

Sample Program

This FastAPI app creates a WebSocket endpoint at '/ws'. When clients connect, they are added to the manager. When a client sends a message, it is broadcast to all connected clients. If a client disconnects, it is removed.

FastAPI

from fastapi import FastAPI, WebSocket
from fastapi.websockets import WebSocketDisconnect

app = FastAPI()

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()

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f'Message: {data}')
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Important Notes

Make sure to handle client disconnects to avoid errors.

Broadcasting works well for chat apps, live updates, and notifications.

Test with multiple clients connected to see messages sent to all.

Summary

Broadcasting sends the same message to many clients at once.

Use a connection manager to track and send messages to all clients.

FastAPI WebSockets make it easy to build real-time broadcast features.