[Fastapi] 웹소켓 구축

개요

4-2 캡스톤 디자인 프로젝트에서 제공하는 웹 서비스에서 가장 중요한 것은 실시간 업데이트이다.

 

여러 해결 방식이 있겠지만 websocket을 통해 실시간 업데이트를 구현하였다.

 

React와 S3를 웹소켓으로 양방향 연결을 하고 용도에 따라 여러 개의 트리거로 나눠 연결된 웹소켓을 활용하였다.

 

작동 방식은 동일하다. 특정 작업이 끝내면 사용자에게 실시간으로  알림을 준다. 

 

코드

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware

router = APIRouter()

# Connection 관리
class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

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

    async def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

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

manager = ConnectionManager()

@router.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        await manager.disconnect(websocket)

async def notify_clients(message: str):
    await manager.broadcast(message)

 

각각 모듈화된 라우터에서 해당 웹소켓 라우터를 import한 뒤 notify_clients를 호출해서 사용한다.

결과

 

Fast api log와 Chrome 개발자 도구를 통해 웹소켓이 정상 연결된 것을 확인 가능하다.