Блог AST-SoftPro

WebSocket в Python: реальный-time уведомления и чаты

28.05.2026 12 мин чтения

Введение в WebSocket и real-time коммуникации

WebSocket — это протокол передачи данных между клиентом и сервером, обеспечивающий двустороннюю связь по одному TCP-соединению. В отличие от HTTP (который работает по модели запрос-ответ), WebSocket позволяет отправлять данные без явного запроса клиента, что делает его идеальным для приложений с real-time функционалом: чатов, уведомлений, онлайн-мониторинга и других сценариев.

Python предоставляет несколько библиотек для работы с WebSocket, но наиболее популярной на сегодняшний день остаётся websockets (для асинхронных серверов) и интеграция через FastAPI. FastAPI — это фреймворк для создания API в Python, который поддерживает WebSocket нативно благодаря асинхронной архитектуре. Это позволяет эффективно обрабатывать множество одновременных подключений без значительных затрат ресурсов.

В данной статье рассматриваются основы реализации real-time систем на Python с использованием WebSockets и FastAPI: создание сервера, управление соединениями, рассылка сообщений (broadcast) всем или группе пользователей, обработка ошибок и безопасность соединений.

Установка зависимостей

Для начала необходимо установить необходимые библиотеки:

pip install fastapi uvicorn websockets python-multipart

FastAPI используется как основной фреймворк, uvicorn — ASGI-сервер для запуска приложения (поддерживает WebSocket), websockets может использоваться в дополнении или заменяться через встроенный механизм FastAPI.

Создание базового WebSocket-сервера на FastAPI

Пример простого сервера с возможностью отправки сообщений всем подключённым клиентам:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

manager = ConnectionManager()


class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, connection: WebSocket):
        await connection.accept()
        self.active_connections.append(connection)

    async def disconnect(self, connection: WebSocket):
        if connection in self.active_connections:
            self.active_connections.remove(connection)

    async def send_message(self, message: str, except_self=None):
        for conn in self.active_connections:
            # Исключаем отправителя
            if conn != except_self:
                await conn.send_text(message)


@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)

    try:
        while True:
            data = await websocket.receive_text()
            # Рассылаем сообщение всем подключённым
            await manager.send_message(data, except_self=websocket)
    except WebSocketDisconnect:
        await manager.disconnect(websocket)

Здесь определён класс ConnectionManager, управляющий подключениями. Каждый новый клиент проходит через /ws эндпоинт, который добавляет его в список активных соединений и начинает принимать сообщения.

Управление соединениями: подключение, отключение и рассылка

WebSocket-соединение может быть открыто, закрыто или разорвано (например, из-за таймаута). Важно корректно обрабатывать все три состояния:

websocket.open — соединение активно.
websocket.closed — клиент отключился добровольно (close()).
WebSocketDisconnect — исключение при потере соединения (сетевые ошибки и т.п.).

В примере выше используется кастомный ConnectionManager, который хранит множество активных WebSocket объектов. При получении сообщения от любого клиента, оно рассылается всем остальным.

Для тестирования можно использовать простой JavaScript-клиент:

const ws = new WebSocket('ws://localhost:8000/ws');

ws.onopen = () => {
    console.log('Соединение установлено');
};

ws.onmessage = (event) => {
    document.getElementById('messages').appendChild(
        createElementWithMessage(event.data)
    );
};

function sendMessage() {
    const input = document.getElementById('input');
    ws.send(input.value);
    input.value = '';
}

Групповая рассылка и исключения из списка получателей

Часто требуется не отправлять сообщение всем, а только определённой группе. Для этого можно расширить ConnectionManager:

from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[str, set[WebSocket]] = {}  # room_name -> connections
        self.global_connected: set[WebSocket] = set()

    async def connect_room(self, room: str, connection: WebSocket):
        if room not in self.active_connections:
            self.active_connections[room] = set()
        self.active_connections[room].add(connection)
        self.global_connected.add(connection)

    async def disconnect_from_room(self, room: str, connection: WebSocket):
        if room in self.active_connections:
            self.active_connections[room].discard(connection)  # безопасно
        self.global_connected.discard(connection)

    async def send_to_room(self, room: str, message: str, except_self=None):
        connections = self.active_connections.get(room, set())
        for conn in connections:
            if conn != except_self and not conn.client_disconnected:
                try:
                    await conn.send_text(message)
                except Exception as e:
                    print(f'Ошибка отправки: {e}')


@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect_room('general', websocket)  # или другое имя комнаты

    try:
        while True:
            data = await websocket.receive_text()
            room, msg = parse_data(data)
            if room == 'broadcast':
                await broadcast(msg)
            else:
                await manager.send_to_room(room, msg, except_self=websocket)
    except WebSocketDisconnect:
        for room in list(manager.active_connections.keys()):
            if websocket in manager.active_connections[room]:
                await manager.disconnect_from_room(room, websocket)

Теперь сообщения могут отправляться в конкретные комнаты (send_to_room) или рассылаться глобально (broadcast). Это полезно для чатов с разделением по темам.

Обработка ошибок и отказоустойчивость

WebSocket соединения подвержены сбоям: медленная сеть, закрытие браузера, таймауты. Важно не допускать падения сервера из-за одного плохого клиента.

В FastAPI при возникновении WebSocketDisconnect или исключений внутри обработчика (await conn.send_text() может выбросить ошибку), приложение должно продолжать работать и обрабатывать другие подключения.

Пример защиты:

async def send_message(self, message: str, except_self=None):
    for conn in self.active_connections:
        try:
            await asyncio.wait_for(conn.send_text(message), timeout=5.0)
        except Exception as e:
            print(f'Не удалось отправить сообщение клиенту {id(conn)}')  # логируем, но не прерываем цикл

Использование asyncio.wait_for() ограничивает время ожидания ответа от клиента и предотвращает зависание.

Также важно обрабатывать случаи, когда клиент отправляет данные без чтения (например, если соединение разорвано). В таких случаях receive_text() выбросит исключение — оно должно быть перехвачено в блоке try/except внутри цикла.

Безопасность WebSocket-соединений

Хотя WebSocket работает поверх HTTPS и защищён на уровне TLS, важно учитывать следующие аспекты:

Аутентификация: клиент должен пройти проверку перед подключением. Это можно реализовать через токен (JWT) или капчу.
Валидация сообщений: все входящие данные должны проходить очистку от HTML/JS-инъекций.
Ограничение частоты запросов (rate limiting): предотвратить DoS-атаки.

Пример проверки JWT при подключении:

from fastapi import Depends, HTTPException
import jwt


async def get_current_user(websocket: WebSocket) -> dict:
    token = websocket.query_params.get('token')
    if not token:
        raise HTTPException(status_code=401, detail='Токен отсутствует')

    try:
        payload = jwt.decode(token, 'secret', algorithms=['HS256'])
        return {'user_id': payload['sub']}
    except Exception:
        raise HTTPException(status_code=403, detail='Недействительный токен')


@app.websocket('/ws/auth')
async def websocket_endpoint_with_auth(
    websocket: WebSocket,
    user: dict = Depends(get_current_user)
):
    await manager.connect(user['user_id'], websocket)

Теперь доступ к /ws/auth возможен только после успешной проверки токена.

Производительность и масштабируемость

FastAPI + Uvicorn — один из самых быстрых Python-фреймворков для ASGI-приложений. При правильной архитектуре он способен обрабатывать тысячи одновременных WebSocket-соединений на одной машине (в зависимости от нагрузки).

Для повышения производительности:

Используйте Gunicorn с Workers (--worker-class uvicorn.workers.UvicornWorker) — позволяет запускать несколько воркеров.
Для высокой нагрузке рассмотрите Redis + Pub/Sub или RabbitMQ для рассылки сообщений, а WebSocket использовать только как канал доставки (push).

Заключение

WebSocket в Python с использованием FastAPI предоставляет мощный и гибкий инструмент для реализации real-time приложений. Основные принципы:

Управление соединениями через ConnectionManager.
Обработка ошибок и исключений — ключ к отказоустойчивости.
Безопасность: аутентификация, валидация данных, rate limiting.
Масштабируемость возможна за счёт правильной архитектуры (Redis, воркеры).

Реализация чатов, уведомлений и других real-time сервисов становится простой задачей при использовании современных подходов. Важно помнить о надёжности соединения, защите от ошибок и возможной нагрузке.

Дополнительные ресурсы

FastAPI WebSocket Documentation — официальная документация.
websockets library: https://github.com/aaugustin/websockets
GitHub Gist с примерами: https://gist.github.com/tiangolo/d542d67085139cddcfb9a090e47a9fcb

Другие записи блога

Все статьи

К списку статей