В прошлом шаге мы реализовали систему WebSocket, она открывает подключение для каждого пользователя, попробуем связать эти подключения, реализовав простую систему наподобие чата. Создадим новое приложение и по аналогии с прошлым шагом создадим конечную точку, выводящую HTML шаблон:

from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
from fastapi import FastAPI


app = FastAPI()
templates = Jinja2Templates(directory="templates")


@app.get("/", response_class=HTMLResponse)
def read_index(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

Как и раньше, будем отправлять вызовы веб-сокетов на основе JS из интерфейса в бэкэнд. Давайте сначала перейдем к бэкэнду. Продолжим написание кода:

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


class ConnectionManager:
    def __init__(self):
        self.connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.connections.append(websocket)

    async def broadcast(self, data: str):
        for connection in self.connections:
            await connection.send_text(data)

Класс ConnectionManager выступает в качестве центрального центра для обработки подключений WebSocket. У него есть метод подключения и отправки сообщений подключенным клиентам.

При создании экземпляра ConnectionManager он инициализирует пустой список под названием connections для отслеживания подключенных пользователей.

Метод подключения(connect) класса ConnectionManager вызывается при установлении нового соединения WebSocket. В качестве параметров принимает экземпляр websocket. Этот метод принимает новое соединение, отправляя сигнал принятия пользователю с помощью вызова await websocket.accept(). Затем он добавляет новый экземпляр WebSocket в список активных соединений.

Метод broadcast предназначен для отправки сообщения всем подключенным пользователям. Он принимает сообщение и экземпляр веб-сокета отправителя. Он итерирует через активные соединения и отправляет сообщение каждому подключенному пользователю.

Чтобы было легче понять работу данного менеджера, мы можем выводить в консоль следующие сообщения: 

class ConnectionManager:
    def __init__(self):
        self.connections: List[WebSocket] = []
        print("Creating a list to active connections", self.connections)

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.connections.append(websocket)
        print("New Active connections are ", self.connections)

    async def broadcast(self, data: str):
        for connection in self.connections:
            await connection.send_text(data)
            print("In broadcast: sent msg to ", connection)

Продолжим написание кода:

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


manager = ConnectionManager()


@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client {client_id}: {data}")
    except WebSocketDisconnect as e:
        manager.connections.remove(websocket)
        print(f'Connection closed {e.code}')

Каждый пользователь будет подключаться к http://127.0.0.1:8000/ws/{client_id}, где {client_id} будет числом, идентифицирующим этого пользователя. После подключения он начнет принимать сообщения из браузера. Любое из этих сообщений будет транслироваться любому пользователю в чате.

В следующем шаге напишем шаблон с JS кодом и протестируем работу.