FastAPI와 WebSockets의 흥미로운 세계에 오신 것을 환영합니다! 오늘은 FastAPI를 사용하여 WebSockets의 메커니즘을 깊이 탐구할 것입니다. 숙련된 개발자이든 방금 시작한 초보자이든 이 가이드는 WebSockets의 필수 개념과 실용적인 구현 방법을 안내합니다. 하지만 시작하기 전에 API 개발을 간편하게 만들어 줄 유용한 도구인 Apidog를 잊지 말아주세요. Apidog는 무료로 다운로드할 수 있으며 API 테스트와 문서를 손쉽게 관리할 수 있도록 도와줍니다.
FastAPI란 무엇인가요?
FastAPI는 표준 Python 타입 힌트를 기반으로 Python 3.7+로 API를 구축하기 위한 현대적이고 빠른(고성능) 웹 프레임워크입니다. 이를 통해 안정적이고 성능이 뛰어난 웹 API를 신속하게 생성할 수 있습니다. FastAPI의 두드러진 특징 중 하나는 비동기 프로그래밍 및 WebSockets를 손쉽게 처리할 수 있는 능력입니다.
WebSockets란 무엇인가요?
WebSockets는 단일 TCP 연결을 통해 전이중 통신 채널을 제공하는 통신 프로토콜입니다. 요청-응답 프로토콜인 HTTP와 달리 WebSockets는 클라이언트와 서버 간의 양방향 통신을 가능하게 합니다. 이는 채팅 애플리케이션, 실시간 알림 및 게임과 같은 실시간 애플리케이션에 적합합니다.
왜 FastAPI와 WebSockets를 함께 사용해야 하나요?
FastAPI의 설계는 WebSockets를 지원하는 API를 쉽게 생성할 수 있게 합니다. FastAPI를 사용하면 빠르고 신뢰할 수 있을 뿐만 아니라 WebSockets를 사용한 비동기 통신을 지원하는 API를 만들 수 있습니다. 이 조합은 현대의 실시간 애플리케이션을 개발하는 데 완벽합니다.
FastAPI WebSockets 시작하기
시작하려면 FastAPI와 FastAPI 애플리케이션을 제공하기 위한 ASGI 서버인 Uvicorn을 설치해야 합니다.
pip install fastapi uvicorn
필요한 패키지를 설치한 후에는 WebSocket 지원이 포함된 FastAPI 애플리케이션을 만들기 시작할 수 있습니다.
기본 FastAPI WebSocket 애플리케이션 만들기
FastAPI에서 WebSocket 통신의 기본 개념을 이해하기 위해 간단한 예제부터 시작해 보겠습니다.
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>WebSocket 예제</title>
</head>
<body>
<h1>WebSocket 예제</h1>
<button onclick="connectWebSocket()">연결</button>
<script>
function connectWebSocket() {
const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
const message = event.data;
alert("서버에서 온 메시지: " + message);
};
ws.onopen = function() {
ws.send("안녕하세요, 서버");
};
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
data = await websocket.receive_text()
await websocket.send_text(f"메시지 텍스트는: {data}")
이 예제에서는 버튼을 클릭할 때 WebSocket 서버에 연결되는 간단한 HTML 페이지가 있습니다. 서버는 "안녕하세요, 서버"라는 메시지를 받고 "메시지 텍스트는: 안녕하세요, 서버"라는 응답을 보냅니다.
여러 연결 처리하기
WebSockets의 강점 중 하나는 여러 클라이언트 연결을 동시에 처리할 수 있다는 것입니다. 이제 우리의 예제를 확장하여 여러 연결을 처리하고 모든 연결된 클라이언트에게 메시지를 방송해 보겠습니다.
from fastapi import FastAPI, WebSocket
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 broadcast(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.broadcast(f"클라이언트가 말했습니다: {data}")
except:
manager.disconnect(websocket)
이 업데이트된 예제에서는 여러 WebSocket 연결을 관리하기 위해 ConnectionManager 클래스를 도입했습니다. broadcast 메서드는 모든 연결된 클라이언트에게 메시지를 전송합니다. 이렇게 하면 모든 클라이언트가 다른 클라이언트가 보낸 메시지를 받을 수 있습니다.
WebSocket 엔드포인트에 매개변수 전달하기
때때로 WebSocket 엔드포인트에 매개변수를 전달해야 할 수도 있습니다. FastAPI를 사용하면 이를 쉽게 수행할 수 있습니다. 다음은 예입니다:
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
await websocket.accept()
await websocket.send_text(f"안녕하세요, 클라이언트 {client_id}")
while True:
data = await websocket.receive_text()
await websocket.send_text(f"{client_id}에서 온 메시지: {data}")
이 예제에서는 client_id 매개변수를 받는 WebSocket 엔드포인트를 정의합니다. 이 매개변수는 클라이언트를 식별하고 메시지를 개인화하는 데 사용됩니다.
API 엔드포인트와 WebSockets 통합하기
FastAPI는 기존 API 엔드포인트와 WebSocket 통신을 통합할 수 있도록 해줍니다. 이는 실시간 통신과 전통적인 API 요청을 결합해야 할 경우 유용합니다.
from fastapi import FastAPI, WebSocket, HTTPException
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id == 0:
raise HTTPException(status_code=404, detail="항목을 찾을 수 없습니다")
return {"item_id": item_id, "name": "항목 이름"}
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
await websocket.send_text("WebSocket에 연결되었습니다")
while True:
data = await websocket.receive_text()
await websocket.send_text(f"수신됨: {data}")
이 예제에서는 전통적인 API 엔드포인트 /items/{item_id}와 WebSocket 엔드포인트 /ws가 함께 있습니다. 두 엔드포인트는 동시에 사용 가능하며, 이를 통해 애플리케이션에서 동기 및 비동기 통신을 모두 처리할 수 있습니다.
보안 고려 사항
WebSockets를 사용할 때는 인증 및 권한 부여와 같은 보안 측면을 고려하는 것이 중요합니다. FastAPI는 WebSocket 연결을 보호하기 위한 도구를 제공합니다.
WebSocket 연결 보안 강화하기
FastAPI의 의존성 주입 시스템을 사용하여 WebSocket 엔드포인트를 보호할 수 있습니다. 다음은 간단한 토큰 기반 인증으로 WebSocket 엔드포인트를 보호하는 방법의 예입니다.
from fastapi import FastAPI, WebSocket, Depends, HTTPException, status
app = FastAPI()
def get_token_header(token: str = None):
if token != "mysecrettoken":
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="유효하지 않은 토큰")
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_token_header)):
await websocket.accept()
await websocket.send_text("WebSocket에 연결되었습니다")
while True:
data = await websocket.receive_text()
await websocket.send_text(f"수신됨: {data}")
이 예제에서는 Depends 함수를 사용하여 get_token_header 의존성을 주입합니다. 이 함수는 토큰의 유효성을 검사합니다. 토큰이 유효하지 않으면 HTTP 403 Forbidden 오류가 발생합니다.
바이너리 데이터 처리하기
WebSockets는 바이너리 데이터 전송도 지원합니다. 이는 파일이나 이미지를 전송하는 데 유용할 수 있습니다. FastAPI는 바이너리 데이터를 쉽게 처리할 수 있게 해줍니다.
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
await websocket.send_text("WebSocket에 연결되었습니다")
while True:
data = await websocket.receive_bytes()
await websocket.send_bytes(data)
이 예제에서 WebSocket 엔드포인트는 receive_bytes를 사용하여 바이너리 데이터를 수신하고 send_bytes를 사용하여 클라이언트에게 다시 전송합니다.
FastAPI WebSockets의 실제 응용 프로그램
WebSockets는 매우 다재다능하며 다양한 실제 응용 프로그램에 사용될 수 있습니다. 다음은 몇 가지 예입니다:
실시간 채팅 애플리케이션
WebSockets의 가장 일반적인 사용 사례 중 하나는 실시간 채팅 애플리케이션을 구축하는 것입니다. FastAPI를 사용하면 여러 클라이언트를 처리하고 메시지를 실시간으로 방송하는 채팅 서버를 쉽게 만들 수 있습니다.
실시간 알림
WebSockets는 실시간 알림을 제공하는 데 적합합니다. 예를 들어, WebSockets를 사용하여 새 메시지, 업데이트 또는 알림을 사용자에게 실시간으로 알릴 수 있습니다.
온라인 게임
WebSockets는 온라인 게임에서 플레이어 간의 실시간 통신을 가능하게 하는 데 널리 사용됩니다. FastAPI의 비동기 프로그래밍 지원은 게임 서버를 구축하는 데 매우 유용합니다.
WebSocket 클라이언트를 디버깅하기 위해 Apidog를 사용하는 방법은?
API와 실시간 데이터를 관리하는 것은 어려울 수 있습니다. Apidog가 바로 그 해결책입니다. Apidog는 API 개발, 테스트 및 관리를 간소화하여 FastAPI와 WebSockets를 사용하여 HTTP 요청과 실시간 업데이트를 모두 처리하는 데 더 쉽게 해줍니다.
- Apidog 열기: 먼저 Apidog 애플리케이션을 시작하고 왼쪽에 있는 "+" 버튼을 클릭합니다. 그러면 새로운 드롭다운이 열립니다. 이곳에서 "새 WebSocket API":를 선택합니다.
2. 연결 설정하기: Apidog의 주소 표시줄에 WebSocket API URL을 입력하여 시작합니다. 그런 다음 "연결" 버튼을 클릭하여 핸드셰이크 프로세스를 시작하고 연결을 설정할 수 있습니다. Apidog는 핸드셰이크 중에 매개변수인 Params, Headers 및 Cookies를 사용자 지정할 수 있도록 해줍니다.
3. 메시지 송수신하기: 연결이 설정되면 "메시지" 탭에서 메시지를 보낼 수 있습니다. 텍스트, JSON, XML, HTML 및 기타 텍스트 형식의 메시지뿐만 아니라 Base64 또는 16진수 형식의 바이너리 형식 메시지를 작성할 수 있습니다. Apidog의 새로운 타임라인 뷰는 연결 상태, 송신된 메시지 및 수신된 메시지를 시간 순서대로 보여줍니다. 메시지를 클릭하면 메시지의 세부정보를 쉽게 확인할 수 있습니다.
4. API 문서화하기: Apidog는 WebSocket API에 대한 강력한 API 문서화 기능을 상속받아 WebSocket 상호작용을 효과적으로 문서화할 수 있습니다.
디버깅 과정 중에 연결 문제, 메시지 형식 오류 및 보안 문제와 같은 일반적인 WebSocket 문제를 확인하는 것을 잊지 마세요.
HTTP 요청을 보내기 위해 Apidog 사용하기
Apidog는 HTTP 요청을 테스트하는 기능을 더욱 향상시키는 여러 고급 기능을 제공합니다. 이러한 기능을 통해 요청을 사용자 지정하고 보다 복잡한 시나리오를 손쉽게 처리할 수 있습니다.
1단계: Apidog를 열고 새 요청을 만듭니다.
2단계: 만들고자 하는 POST 요청의 API 세부정보를 찾거나 직접 입력합니다.
3단계: 요청 본문에 포함하려는 필수 매개변수 및 데이터를 입력합니다.
결론
FastAPI와 WebSockets는 현대의 실시간 애플리케이션을 구축하는 데 강력한 조합입니다. 채팅 애플리케이션, 실시간 알림 시스템 또는 온라인 게임을 구축하든, FastAPI는 WebSocket 통신을 효율적으로 처리하는 데 필요한 도구를 제공합니다.
API 개발 과정을 간소화하기 위해 Apidog를 무료로 다운로드하는 것을 잊지 마세요. Apidog는 API를 테스트하고 문서화하는 과정을 쉽게 만들어 시간을 절약할 수 있도록 합니다.
