Как реализовать WebSocket в FastAPI?

Что такое WebSocket

WebSocket — это постоянное двунаправленное соединение между клиентом и сервером. В отличие от обычного HTTP-запроса, сервер может отправлять сообщения клиенту без нового запроса от браузера.

Простой endpoint

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            message = await websocket.receive_text()
            await websocket.send_text(f'Received: {message}')
    except WebSocketDisconnect:
        pass

Менеджер подключений

Для чатов и подписок обычно создают connection manager, который хранит активные подключения и умеет отправлять сообщения одному клиенту, комнате или всем клиентам текущего процесса.

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)

Аутентификация и авторизация

Проверку токена можно выполнить до accept(), передав его в query parameter, cookie или согласованный header. После аутентификации нужно проверить, имеет ли пользователь доступ к конкретному каналу, комнате или объекту, а не только валидность токена.

Масштабирование

Список подключений в памяти доступен только одному worker-процессу. При нескольких workers или репликах сообщения нужно распространять через Redis Pub/Sub, NATS, Kafka, RabbitMQ или специализированный realtime-сервис. Балансировщик также должен поддерживать WebSocket upgrade и корректные тайм-ауты.

Надёжность

Нужно обрабатывать отключения, heartbeat или ping/pong, лимиты размера сообщений, backpressure и повторное подключение клиента. Долгие или CPU-bound действия не выполняют внутри цикла WebSocket: их выносят в фоновые workers.

Как ответить на собеседовании

В FastAPI WebSocket создаю через @app.websocket(), принимаю соединение методом accept() и обрабатываю сообщения в цикле. Для одного процесса можно хранить подключения в connection manager, но для нескольких реплик обязательно нужен общий Pub/Sub или broker, иначе сообщения не попадут клиентам на других workers.