FastAPI를 사용하여 인스턴트 메시징 서비스 구현하기

25 Sep 2023

FastAPI

인스턴트 메시징 서비스는 실시간 메시지 송수신을 지원하는 플랫폼입니다. 이번 글에서는 FastAPI를 사용하여 인스턴트 메시징 서비스를 구현하는 방법에 대해 알아보겠습니다.

FastAPI란?

FastAPI는 Python 기반의 빠르고 현대적인 웹 프레임워크입니다. ASGI(Asynchronous Server Gateway Interface)를 사용하여 비동기 및 실시간 기능을 지원하며, 자동으로 API 문서를 생성하는 등 개발자 편의 기능이 많이 포함되어 있습니다.

프로젝트 설정

우선, FastAPI를 사용하기 위해 프로젝트를 설정해야 합니다. 가상환경을 만들고 FastAPI를 설치합니다. 다음 명령을 터미널에서 실행하세요.

pip install fastapi

프로젝트 디렉토리를 생성하고 main.py 파일을 만들어주세요.

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

인스턴트 메시징 기능 추가하기

이제 기본적인 FastAPI 애플리케이션을 구성했으므로, 인스턴트 메시징 기능을 추가해보겠습니다. 메시지는 WebSocket을 통해 실시간으로 송수신되므로, FastAPI의 WebSocket 기능을 사용하겠습니다.

from fastapi import FastAPI, WebSocket

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections = []

    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_message(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)

    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_message(data)
    except Exception as e:
        print(e)
        manager.disconnect(websocket)

위 코드에서 ConnectionManager 클래스는 활성 WebSocket 연결을 관리하는 역할을 합니다. /ws 엔드포인트는 클라이언트와 WebSocket 연결을 설정하고, 메시지 수신 및 송신을 처리합니다.

실행 및 테스트

이제 FastAPI 애플리케이션을 실행하고 테스트해보겠습니다. 다음 명령을 터미널에서 실행하세요.

uvicorn main:app --reload

브라우저나 API 클라이언트 도구에서 http://localhost:8000/에 접속하면 “Hello, World!” 메시지가 반환됩니다.

WebSocket 기능을 테스트하기 위해 WebSocket 클라이언트를 사용해보세요.

import websockets
import asyncio

async def send_message():
    async with websockets.connect("ws://localhost:8000/ws") as websocket:
        await websocket.send("Hello, Server!")
        response = await websocket.recv()
        print(f"Received: {response}")

asyncio.get_event_loop().run_until_complete(send_message())

위 코드를 실행하면 WebSocket 서버에 “Hello, Server!” 메시지가 전송되고, 서버로부터 응답을 받아 출력합니다.

결론

FastAPI를 사용하여 인스턴트 메시징 서비스를 구현하는 방법에 대해 알아보았습니다. FastAPI의 강력한 기능과 WebSocket을 통한 실시간 통신을 통해 효율적이고 빠른 인스턴트 메시징 서비스를 구축할 수 있습니다.

#FastAPI #인스턴트메시징