websocket 설치

설치는 다음 커멘드로 설치한다.

pip install websocket

websocket 서버 구축

fastapi는 websocket API를 지원하고 있다. 즉, 브라우저 간의 양방향 통신을 지원한다고 할 수 있다.

fastapi에서 websocket을 사용하려면 먼저 websocket 모듈을 임포트해주어야한다.

from fastapi import WebSocket

websocket 모듈을 임포트했다면, 다음과 같이 라우팅을 지정할 수 있다.

코드는 fastapi에서 제공하는 예제 코드를 가져왔다.

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

socket 통신의 과정을 생각하면 위의 코드가 쉽게 보일 것이다.

Untitled

서버는 fastapi를 통해 websocket을 만들어 항상 listen상태로 있는다. 여기서 클라이언트를 통해 connect요청이 들어오면 accept을 해주고 메세지를 서로 주고 받는 형태로 이어진다.

이 accept 과정이 await websocket.accept() 코드이다.

이후 무한반복문을 통해 클라이언트에서 온 메세지를 다시 전달해주는 코드이다.

<aside> 💡 websocket 접속 시, 일반적인 웹 프로토콜과 다르므로, ws라는 프로토콜을 사용해야한다. 즉, ‘http://127.0.0.1:8000’ 가 아닌 ‘ws://127.0.0.1:8000’ 가 되어야한다.

</aside>

다중 커넥션 관리

먼저 여러 클라이언트로부터 오는 커넥션을 관리하기 위해 다음과 같이 ConnectionManager 유틸 클래스를 만들어준다.

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 send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)