FastAPI와 웹소켓을 이용한 Python 실시간 통신

소개

오늘날 상호 연결된 세상에서는 즉각적인 업데이트와 동적인 사용자 경험에 대한 요구가 그 어느 때보다 높습니다. 협업 문서 편집부터 라이브 채팅 애플리케이션에 이르기까지 실시간 통신 능력은 현대 웹 애플리케이션에 매우 중요합니다. 많은 작업을 위해 강력한 HTTP 요청-응답 모델도 지속적인 양방향 통신이 필요할 때는 종종 부족합니다. 바로 이때 웹소켓과 같은 기술이 등장하며, 서버가 정보를 가능한 한 빨리 클라이언트로 푸시하고 그 반대도 가능하게 하는 지속적인 연결을 제공합니다. Python은 강력한 생태계와 프레임워크를 통해 이러한 실시간 시스템을 구축하기 위한 훌륭한 도구를 제공합니다. 이 글에서는 Python에서 실시간 통신을 구현하는 방법을 살펴보고, 특히 웹소켓을 활용하며, FastAPI가 이 프로세스를 단순화하여 개발자가 매우 상호 작용적인 애플리케이션을 쉽게 만들 수 있도록 하는 방법을 시연할 것입니다.

실시간 통신의 핵심 개념

구현 세부 사항을 자세히 알아보기 전에 Python 맥락에서 실시간 통신을 뒷받침하는 핵심 개념을 이해하는 것이 중요합니다.

웹소켓: 웹소켓은 단일의 오래 지속되는 TCP 연결을 통해 전이중(full-duplex) 통신 채널을 제공합니다. 상태 비저장(stateless)이고 일반적으로 각 요청 후에 연결을 닫는 HTTP와 달리, 웹소켓 연결은 열린 상태를 유지하여 클라이언트와 서버 간의 지속적인 양방향 데이터 교환을 가능하게 합니다. 이는 전통적인 폴링 방식에 비해 지연 시간과 오버헤드를 크게 줄입니다.

FastAPI: FastAPI는 Python 3.7+을 기반으로 표준 Python 타입 힌트를 사용하여 API를 구축하기 위한 현대적이고 빠른(고성능) 웹 프레임워크입니다. Starlette(웹 부분)과 Pydantic(데이터 유효성 검사 및 직렬화)을 기반으로 구축되었습니다. FastAPI는 본질적으로 비동기 프로그래밍(async/await)을 지원하므로 동시 웹소켓 연결을 효율적으로 처리하는 데 탁월한 선택입니다.

비동기 프로그래밍 (async/await): Python의 asyncio 라이브러리와 async/await 구문은 주 스레드를 차단하지 않고 여러 동시 웹소켓 연결을 처리하는 데 기본입니다. 이를 통해 애플리케이션은 수많은 클라이언트를 동시에 관리하여 응답성과 확장성을 보장할 수 있습니다.

연결 관리: 실시간 애플리케이션에서는 서버가 활성 웹소켓 연결을 관리해야 합니다. 클라이언트가 연결되면 해당 웹소켓 객체는 일반적으로 서버가 특정 클라이언트로 메시지를 보내거나 모든 연결된 클라이언트로 브로드캐스트할 수 있도록 저장됩니다(예: 목록 또는 사전). 마찬가지로 서버는 비활성 클라이언트를 제거하기 위해 연결 끊김 이벤트를 처리해야 합니다.

실시간 통신 구현

FastAPI와 웹소켓을 사용하여 기본적인 실시간 통신 시스템을 구현하는 방법을 살펴보겠습니다. 예시로 간단한 채팅 애플리케이션을 구축할 것입니다.

1. FastAPI 설정

먼저 FastAPI와 Uvicorn(ASGI 서버)이 설치되어 있는지 확인하세요:

pip install fastapi uvicorn websockets

2. 기본 웹소켓 엔드포인트

FastAPI에서 기본 웹소켓 엔드포인트를 정의하는 방법은 다음과 같습니다:

# main.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

# 활성 웹소켓 연결 저장
websocket_connections: List[WebSocket] = []

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()  # 웹소켓 연결 수락
    websocket_connections.append(websocket)
    try:
        while True:
            data = await websocket.receive_text()  # 클라이언트로부터 메시지 수신
            print(f"Received message: {data}")
            # 보낸 사람에게 메시지 에코
            await websocket.send_text(f"Message text was: {data}")

            # 선택 사항: 모든 연결된 클라이언트로 브로드캐스트
            # for connection in websocket_connections:
            #     if connection != websocket: # 보낸 사람에게 다시 보내지 않음
            #         await connection.send_text(f"Someone said: {data}")
    except WebSocketDisconnect:
        websocket_connections.remove(websocket)
        print("Client disconnected")

# 애플리케이션 실행 방법: uvicorn main:app --reload

이 예시에서는:

• @app.websocket("/ws")는 웹소켓 경로를 정의합니다.
• websocket_endpoint 함수는 WebSocket 객체를 수락하는 async 함수입니다.
• await websocket.accept()는 웹소켓 연결을 설정합니다.
• while True 루프는 await websocket.receive_text()를 사용하여 들어오는 메시지를 지속적으로 수신 대기합니다.
• await websocket.send_text()는 클라이언트에게 메시지를 다시 보냅니다.
• 클라이언트가 연결을 끊으면 WebSocketDisconnect 예외가 포착되어 websocket_connections 목록을 정리할 수 있습니다.

3. 메시지 브로드캐스트

채팅 애플리케이션의 경우 모든 연결된 클라이언트로 메시지를 브로드캐스트하는 것이 필수적입니다. 이 예시를 향상시켜 이를 지원하도록 하겠습니다:

# main.py (계속)
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

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)

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} says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")

# 테스트를 위해 간단한 HTML 클라이언트 제공
from fastapi.responses import HTMLResponse

@app.get("/")
async def get():
    return HTMLResponse(f"""
        <!DOCTYPE html>
        <html>
            <head>
                <title>Chat App</title>
            </head>
            <body>
                <h1>WebSocket Chat</h1>
                <form action="" onsubmit="sendMessage(event)">
                    <input type="text" id="messageText" autocomplete="off"/>
                    <button>Send</button>
                </form>
                <ul id='messages'>
                </ul>
                <script>
                    var ws = new WebSocket("ws://localhost:8000/ws/1"); // 클라이언트 1로 연결
                    ws.onopen = function(event) {
                        console.log("WebSocket connection opened:", event);
                    };
                    ws.onmessage = function(event) {
                        var messages = document.getElementById('messages')
                        var message = document.createElement('li')
                        var content = document.createTextNode(event.data)
                        message.appendChild(content)
                        messages.appendChild(message)
                    };
                    ws.onclose = function(event) {
                        console.log("WebSocket connection closed:", event);
                    };
                    ws.onerror = function(event) {
                        console.error("WebSocket error observed:", event);
                    };
                    function sendMessage(event) {
                        var input = document.getElementById("messageText")
                        ws.send(input.value)
                        input.value = ''
                        event.preventDefault()
                    }
                </script>
            </body>
        </html>
    """)

이 향상된 버전에서는:

• 활성 연결 관리, 연결, 연결 해제 및 메시지 브로드캐스트 로직을 캡슐화하기 위해 ConnectionManager 클래스를 도입했습니다. 이는 더 깨끗한 코드와 더 나은 조직을 촉진합니다.
• 웹소켓 엔드포인트는 이제 client_id 경로 매개변수를 받으며, 이는 클라이언트를 식별하는 데 사용될 수 있습니다.
• 메시지가 수신되면 manager.broadcast()는 모든 연결된 클라이언트로 전송합니다.
• 채팅 애플리케이션을 브라우저에서 직접 테스트하기 위한 JavaScript가 포함된 간단한 HTML 페이지가 제공됩니다.

애플리케이션 시나리오

웹소켓의 강력함은 수많은 실시간 애플리케이션 시나리오로 확장됩니다.

• 라이브 채팅 애플리케이션: 설명된 대로 사용자와 사용자 간의 실시간 메시지 교환.
• 멀티플레이어 게임: 클라이언트 간의 게임 상태 및 플레이어 행동 동기화.
• 협업 편집: 여러 사용자가 문서를 동시에 편집하는 Google Docs 스타일 애플리케이션.
• 대시보드 업데이트: 대시보드로 푸시되는 실시간 메트릭, 주가 또는 센서 데이터.
• 알림: 사용자가 새로고침할 필요 없이 즉시 경고 및 알림.
• IoT 장치 통신: 명령 및 제어 또는 데이터 스트리밍을 위한 IoT 장치와의 양방향 통신.

결론

실시간 통신은 현대적인 상호 작용 애플리케이션의 초석이며, Python, 특히 FastAPI와 같은 프레임워크는 웹소켓을 사용하여 이를 구현하는 강력하고 효율적인 방법을 제공합니다. 지속적인 전이중 연결을 설정함으로써 웹소켓은 지연 시간을 크게 줄이고 기존 폴링보다 우수한 사용자 경험을 제공합니다. FastAPI의 비동기 특성과 명확한 API 디자인은 이러한 실시간 시스템을 구축하고 확장하는 것을 간단하게 만들어 개발자가 동적이고 반응성이 뛰어난 애플리케이션을 쉽게 만들 수 있도록 합니다. FastAPI와 함께 웹소켓을 마스터하면 Python으로 제공되는 모든 웹 서비스에 대한 상호 작용의 새로운 차원을 열 수 있습니다.