FastAPI로 실시간 채팅 애플리케이션 구축하기

fastapi diagram

FastAPI는 WebSocket을 지원하여 실시간 애플리케이션을 쉽게 개발할 수 있습니다. 이번 포스팅에서는 WebSocket을 활용해 간단한 실시간 채팅 애플리케이션을 구축하는 방법을 단계별로 설명합니다.

WebSocket이란?

WebSocket은 클라이언트와 서버 간 양방향 통신을 가능하게 하는 프로토콜입니다. HTTP와 달리 연결을 유지하며 실시간 데이터 전송에 적합합니다.

• 장점: 낮은 지연 시간, 효율적인 통신, 실시간 업데이트 지원
• 활용 분야: 채팅, 주식 거래, 실시간 알림, 온라인 게임

개발 환경 설정

1. 필요한 라이브러리 설치

FastAPI와 ASGI 서버인 Uvicorn을 설치합니다.

pip install fastapi uvicorn

2. Uvicorn 실행 옵션

• --reload: 코드 변경 시 자동 재시작 (개발용)
• --host, --port: 서버 바인딩 주소와 포트 지정

  uvicorn main:app --reload --host 0.0.0.0 --port 8000

서버 코드 구현 ()

1. 기본 구조 작성

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from typing import List

app = FastAPI()
active_connections: List[WebSocket] = []  # 연결된 클라이언트 관리

2. WebSocket 엔드포인트 설정

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    active_connections.append(websocket)

    try:
        while True:
            data = await websocket.receive_text()
            # 모든 클라이언트에게 메시지 브로드캐스트
            for connection in active_connections:
                await connection.send_text(data)
    except WebSocketDisconnect:
        active_connections.remove(websocket)

3. HTML 클라이언트 제공

@app.get("/")
async def get():
    return HTMLResponse(content=open("index.html").read(), status_code=200)

클라이언트 코드 구현 ()

1. 기본 HTML 구조

    FastAPI 채팅


    실시간 채팅

    전송



        const ws = new WebSocket(`ws://${window.location.host}/ws`);

        // 메시지 수신 처리
        ws.onmessage = (event) => {
            const messages = document.getElementById("messages");
            const li = document.createElement("li");
            li.textContent = event.data;
            messages.appendChild(li);
        };

        // 메시지 전송 처리
        document.getElementById("sendButton").addEventListener("click", () => {
            const input = document.getElementById("messageInput");
            ws.send(input.value);
            input.value = "";
        });

애플리케이션 실행 및 테스트

1. 서버 실행
2. uvicorn main:app --reload
3. 테스트 방법
   • 브라우저에서 http://localhost:8000 접속
   • 여러 탭이나 다른 기기에서 동시 접속 후 메시지 교환 확인

고급 기능 추가하기

1. 비동기 주기적 데이터 전송

import asyncio

@app.websocket("/ws/status")
async def status_updates(websocket: WebSocket):
    await websocket.accept()
    while True:
        await asyncio.sleep(5)
        await websocket.send_text("서버 상태: 정상")

2. 방(Room) 기능 구현

from collections import defaultdict
rooms = defaultdict(list)

@app.websocket("/ws/{room_id}")
async def room_chat(websocket: WebSocket, room_id: str):
    await websocket.accept()
    rooms[room_id].append(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            for conn in rooms[room_id]:
                if conn != websocket:
                    await conn.send_text(data)
    except WebSocketDisconnect:
        rooms[room_id].remove(websocket)

주의사항 및 최적화 팁

1. 연결 관리
   • 클라이언트가 연결을 종료할 때 active_connections에서 제거해야 합니다.
   • try-except 블록으로 WebSocketDisconnect 예외 처리
2. 성능 최적화
   • 대량 메시지 처리 시 await websocket.receive_bytes() 사용
   • Redis나 Pub/Sub 시스템을 활용해 확장성 개선
3. 보안 강화
   • 인증 미들웨어 추가 (예: JWT 토큰 검증)
   • wss:// 프로토콜 사용 (HTTPS 환경)

결론

FastAPI와 WebSocket을 활용하면 실시간 기능이 필요한 애플리케이션을 쉽게 구축할 수 있습니다. 채팅, 실시간 알림, 데이터 모니터링 등 다양한 분야에 적용 가능하며, 비동기 처리 덕분에 높은 성능을 유지할 수 있습니다.
 

[FastAPI] 실시간 애플리케이션 구축

 

FastAPI와 Pydantic: 강력한 데이터 검증과 모델링