WebSockets в FastAPI: чат в реальном времени

Привет! В отличие от HTTP, где клиент отправляет запрос и ждёт ответа, WebSockets позволяют установить постоянное двустороннее соединение. Это идеально для чатов, уведомлений, игр и других приложений реального времени.

В этой статье мы создадим полноценный чат с использованием WebSockets в FastAPI.

В этой статье мы разберём:

Что такое WebSockets и зачем они нужны
Создание WebSocket эндпоинта
Управление подключениями
Отправка и приём сообщений
Комнаты (группы) для приватных чатов
Обработка ошибок

Содержание

Что нужно знать перед началом
Основная часть
Что такое WebSockets
Первый WebSocket эндпоинт
Управление подключениями (менеджер соединений)
Чат с комнатами (группами)
Аутентификация WebSocket
JSON в WebSocket
Ping-Pong (проверка активности)
Полный пример чата
Задачи для закрепления
Нюансы и подводные камни
Обработка отключений
Максимальное количество соединений
Масштабирование
Безопасность
Частые ошибки и как их избежать
Ошибка 1: Забыл await websocket.accept()
Ошибка 2: Не обрабатываешь WebSocketDisconnect
Ошибка 3: Блокирующий код внутри WebSocket
Шпаргалка
Заключение
КВИЗ
Что дальше?

Что нужно знать перед началом

Для этого урока тебе понадобится:

Установленный FastAPI и Uvicorn
Базовое понимание асинхронности

Совет: WebSockets создают постоянное соединение между клиентом и сервером, позволяя обмениваться данными в реальном времени.

Основная часть

Что такое WebSockets

WebSockets — это протокол для постоянного двустороннего обмена данными между клиентом и сервером.

Отличия от HTTP:

Характеристика	HTTP	WebSocket
Тип соединения	Запрос-ответ	Постоянное
Двустороннее	Нет (клиент → сервер)	Да
Реальное время	Нет	Да
Заголовки	Каждый запрос	Один раз при установке

Совет: WebSockets используют те же порты, что и HTTP (80/443), но протокол другой: ws:// и wss://.

Первый WebSocket эндпоинт

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Сообщение получено: {data}")
    except WebSocketDisconnect:
        print("Клиент отключился")

Клиент (JavaScript):

const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = (event) => console.log(event.data);
ws.send("Привет, сервер!");

Совет: WebSocketDisconnect — исключение, которое возникает при отключении клиента.

Управление подключениями (менеджер соединений)

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

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"Сообщение: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("Пользователь покинул чат")

Совет: ConnectionManager — полезный паттерн для управления всеми активными подключениями.

Чат с комнатами (группами)

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

app = FastAPI()

class RoomManager:
    def __init__(self):
        self.rooms: Dict[str, List[WebSocket]] = {}

    async def connect(self, websocket: WebSocket, room: str):
        await websocket.accept()
        if room not in self.rooms:
            self.rooms[room] = []
        self.rooms[room].append(websocket)

    def disconnect(self, websocket: WebSocket, room: str):
        if room in self.rooms:
            self.rooms[room].remove(websocket)
            if not self.rooms[room]:
                del self.rooms[room]

    async def broadcast(self, message: str, room: str):
        if room in self.rooms:
            for connection in self.rooms[room]:
                await connection.send_text(message)

room_manager = RoomManager()

@app.websocket("/ws/{room}")
async def websocket_endpoint(websocket: WebSocket, room: str):
    await room_manager.connect(websocket, room)

    try:
        while True:
            data = await websocket.receive_text()
            await room_manager.broadcast(f"[{room}] {data}", room)
    except WebSocketDisconnect:
        room_manager.disconnect(websocket, room)

Совет: Комнаты позволяют изолировать чаты. Клиенты в одной комнате не видят сообщения из другой.

Аутентификация WebSocket

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
import json

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(
    websocket: WebSocket,
    token: str = Query(...)
):
    # Проверка токена
    if token != "secret":
        await websocket.close(code=1008)
        return

    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Привет, {data}")
    except WebSocketDisconnect:
        print("Клиент отключился")

Клиент (JavaScript):

const ws = new WebSocket("ws://localhost:8000/ws?token=secret");

Совет: Токен можно передавать как query-параметр или в заголовках (но заголовки не поддерживаются в WebSocket API браузера).

JSON в WebSocket

import json
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            try:
                json_data = json.loads(data)
                message = json_data.get("message", "")
                response = {
                    "type": "echo",
                    "received": message,
                    "timestamp": datetime.now().isoformat()
                }
                await websocket.send_text(json.dumps(response))
            except json.JSONDecodeError:
                await websocket.send_text(json.dumps({
                    "error": "Неверный формат JSON"
                }))
    except WebSocketDisconnect:
        print("Клиент отключился")

Клиент (JavaScript):

ws.send(JSON.stringify({ message: "Привет!" }));

Совет: Для структурированных данных используй JSON — это делает обмен данными более удобным.

Ping-Pong (проверка активности)

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            if data == "ping":
                await websocket.send_text("pong")
            else:
                await websocket.send_text(f"Вы сказали: {data}")
    except WebSocketDisconnect:
        print("Клиент отключился")

Совет: Ping-Pong помогает проверять, что соединение всё ещё активно.

Полный пример чата

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict, List
from datetime import datetime
import json

app = FastAPI()

class ChatManager:
    def __init__(self):
        self.rooms: Dict[str, List[WebSocket]] = {}
        self.users: Dict[WebSocket, str] = {}

    async def connect(self, websocket: WebSocket, room: str, username: str):
        await websocket.accept()
        self.users[websocket] = username
        if room not in self.rooms:
            self.rooms[room] = []
        self.rooms[room].append(websocket)
        await self.broadcast(
            json.dumps({"type": "join", "user": username}),
            room
        )

    def disconnect(self, websocket: WebSocket, room: str):
        username = self.users.get(websocket, "Неизвестный")
        if websocket in self.rooms.get(room, []):
            self.rooms[room].remove(websocket)
            if not self.rooms[room]:
                del self.rooms[room]
        if websocket in self.users:
            del self.users[websocket]

    async def broadcast(self, message: str, room: str):
        if room in self.rooms:
            for connection in self.rooms[room]:
                await connection.send_text(message)

    async def send_message(self, websocket: WebSocket, message: str, room: str):
        username = self.users.get(websocket, "Аноним")
        data = {
            "type": "message",
            "user": username,
            "message": message,
            "timestamp": datetime.now().isoformat()
        }
        await self.broadcast(json.dumps(data), room)

chat_manager = ChatManager()

@app.websocket("/ws/{room}")
async def websocket_endpoint(websocket: WebSocket, room: str):
    username = websocket.query_params.get("username", "Аноним")
    await chat_manager.connect(websocket, room, username)

    try:
        while True:
            data = await websocket.receive_text()
            await chat_manager.send_message(websocket, data, room)
    except WebSocketDisconnect:
        chat_manager.disconnect(websocket, room)
        await chat_manager.broadcast(
            json.dumps({"type": "leave", "user": username}),
            room
        )

Задачи для закрепления

Задача 1. Создай WebSocket эндпоинт, который отправляет текущее время каждую секунду.

Задача 2. Добавь аутентификацию по токену.

Задача 3. Реализуй комнаты для разных чатов.

Задача 4. Добавь отправку JSON-сообщений.

Задача 5. Реализуй ping-pong для проверки активности.

Ответы:

Задача 1.

import asyncio
from datetime import datetime

@app.websocket("/ws/time")
async def time_websocket(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            await websocket.send_text(datetime.now().isoformat())
            await asyncio.sleep(1)
    except WebSocketDisconnect:
        pass

Задача 2.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str = Query(...)):
    if token != "secret":
        await websocket.close(code=1008)
        return
    await websocket.accept()

Задача 3.

@app.websocket("/ws/{room}")
async def websocket_endpoint(websocket: WebSocket, room: str):
    # комната доступна через room
    pass

Задача 4.

try:
    json_data = json.loads(data)
    message = json_data.get("message")
except json.JSONDecodeError:
    await websocket.send_text(json.dumps({"error": "Invalid JSON"}))

Задача 5.

if data == "ping":
    await websocket.send_text("pong")

Нюансы и подводные камни

Обработка отключений

Всегда обрабатывай WebSocketDisconnect, чтобы корректно удалять клиента из списка.

Максимальное количество соединений

На сервере есть ограничение на количество одновременных WebSocket-соединений. Используй балансировку для больших проектов.

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

WebSocket-соединения сохраняются на конкретном сервере. Для масштабирования используй Redis или RabbitMQ для синхронизации между серверами.

Безопасность

Используй wss:// (WebSocket over SSL) в продакшене
Валидируй входные данные
Ограничивай размер сообщений
Используй аутентификацию

Частые ошибки и как их избежать

Ошибка 1: Забыл `await websocket.accept()`

Всегда вызывай accept() перед началом работы с WebSocket.

Ошибка 2: Не обрабатываешь `WebSocketDisconnect`

Всегда используй try-except с WebSocketDisconnect.

Ошибка 3: Блокирующий код внутри WebSocket

Используй асинхронные функции (await) для I/O операций.

Шпаргалка

Что нужно	Как пишется
Создать WebSocket эндпоинт	`@app.websocket("/ws")`
Принять соединение	`await websocket.accept()`
Получить сообщение	`await websocket.receive_text()`
Отправить сообщение	`await websocket.send_text(text)`
Обработать отключение	`except WebSocketDisconnect:`
Закрыть соединение	`await websocket.close(code=1000)`

Заключение

Сегодня мы:

Узнали, что такое WebSockets
Создали WebSocket эндпоинт
Реализовали чат с комнатами
Добавили аутентификацию

КВИЗ

Что дальше?

Следующая статья: «Фоновые задачи (BackgroundTasks)»

Что нужно знать перед началом

Основная часть

Что такое WebSockets

Первый WebSocket эндпоинт

Управление подключениями (менеджер соединений)

Чат с комнатами (группами)

Аутентификация WebSocket

JSON в WebSocket

Ping-Pong (проверка активности)

Полный пример чата

Задачи для закрепления

Нюансы и подводные камни

Обработка отключений

Максимальное количество соединений

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

Безопасность

Частые ошибки и как их избежать

Ошибка 1: Забыл await websocket.accept()

Ошибка 2: Не обрабатываешь WebSocketDisconnect

Ошибка 3: Блокирующий код внутри WebSocket

Шпаргалка

Заключение

КВИЗ

Что дальше?

Ошибка 1: Забыл `await websocket.accept()`

Ошибка 2: Не обрабатываешь `WebSocketDisconnect`