[FastAPI] WebSockets

FastAPI WebSockets, FastAPI WebSocket, websockets fastapi, FastAPI 웹소켓, WebSocket broadcast, WebSocket authentication

정의

FastAPI WebSockets 는 Starlette 위에서 양방향 실시간 통신을 제공합니다. HTTP 대신 ws:// / wss:// 스킴을 쓰며, 한 연결 위에서 서버와 클라이언트가 자유롭게 메시지를 주고받습니다.

주요 용도: 채팅, 실시간 알림, 라이브 대시보드, 게임, 협업 편집.

기본 사용

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"echo: {data}")

• accept(): 핸드셰이크 완료, 연결 활성화
• receive_text() / receive_bytes() / receive_json(): 메시지 수신 (blocking, coroutine)
• send_text() / send_bytes() / send_json(): 송신

연결 종료 처리

from fastapi import WebSocketDisconnect

@app.websocket("/ws")
async def ws_ep(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"got: {data}")
    except WebSocketDisconnect:
        print("client disconnected")

클라이언트 종료 -> WebSocketDisconnect 예외. 반드시 catch 하고 자원 정리.

인증 (Query, Cookie, Header)

WebSocket 은 브라우저 API 상 커스텀 헤더를 못 붙임 (subprotocol 은 가능). 대개 query 나 cookie 로 토큰 전달.

Query 파라미터

from fastapi import Query, WebSocketException, status

@app.websocket("/ws")
async def ws_ep(
    websocket: WebSocket,
    token: str = Query(...),
):
    user = validate_token(token)
    if not user:
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
    await websocket.accept()
    ...

Cookie / Header

@app.websocket("/ws")
async def ws_ep(
    websocket: WebSocket,
    session: str | None = Cookie(default=None),
    x_client_id: str | None = Header(default=None),
):
    ...

주의: WebSocket dependency 는 Depends() 도 지원하지만 HTTPException 대신 WebSocketException 을 사용해야 합니다.

Dependency Injection

async def get_current_user_ws(token: str = Query(...)) -> User:
    user = validate_token(token)
    if not user:
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
    return user

@app.websocket("/ws")
async def ws_ep(
    websocket: WebSocket,
    user: Annotated[User, Depends(get_current_user_ws)],
    db: Annotated[AsyncSession, Depends(get_async_db)],
):
    await websocket.accept()
    ...

Broadcast 패턴

여러 클라이언트에 같은 메시지 전송.

단순 in-memory manager

class ConnectionManager:
    def __init__(self):
        self.active: list[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: str):
        for ws in self.active:
            try:
                await ws.send_text(message)
            except Exception:
                pass   # 연결 끊긴 클라이언트

manager = ConnectionManager()

@app.websocket("/chat/{room}")
async def chat(websocket: WebSocket, room: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"[{room}] {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"[{room}] user left")

제한: 단일 프로세스. 여러 워커면 프로세스 간 broadcast 안 됨.

다중 프로세스 broadcast (Redis pub/sub)

import redis.asyncio as redis
import asyncio, json

class BroadcastManager:
    def __init__(self, url: str):
        self.pub = redis.from_url(url)
        self.sub = redis.from_url(url)
        self.local_conns: set[WebSocket] = set()

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.local_conns.add(ws)

    def disconnect(self, ws: WebSocket):
        self.local_conns.discard(ws)

    async def publish(self, channel: str, message: dict):
        await self.pub.publish(channel, json.dumps(message))

    async def subscribe(self, channel: str):
        pubsub = self.sub.pubsub()
        await pubsub.subscribe(channel)
        async for msg in pubsub.listen():
            if msg["type"] != "message":
                continue
            data = json.loads(msg["data"])
            for ws in list(self.local_conns):
                try:
                    await ws.send_json(data)
                except Exception:
                    self.local_conns.discard(ws)

manager = BroadcastManager("redis://localhost")

@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(manager.subscribe("chat"))
    yield
    task.cancel()

app = FastAPI(lifespan=lifespan)

또는 broadcaster 패키지 (encode.io) 로 Redis/Postgres/Memory backend 통합.

Subprotocol

클라이언트가 원하는 서브프로토콜을 협상 (예: MQTT over WebSocket, GraphQL Subscriptions).

@app.websocket("/ws")
async def ws_ep(websocket: WebSocket):
    proto = websocket.headers.get("sec-websocket-protocol", "")
    if "myproto.v1" in proto:
        await websocket.accept(subprotocol="myproto.v1")
    else:
        await websocket.close(code=1002)
        return

클라이언트 예시

const ws = new WebSocket("wss://api.example.com/ws?token=abc");
ws.onopen = () => console.log("open");
ws.onmessage = (e) => console.log("msg:", e.data);
ws.onclose = () => console.log("closed");
ws.send("hello");

Python 클라이언트 (테스트용):

import asyncio
import websockets

async def client():
    async with websockets.connect("ws://localhost:8000/ws?token=abc") as ws:
        await ws.send("hello")
        print(await ws.recv())

asyncio.run(client())

Heartbeat / Ping-Pong

Idle 연결을 감지하려면 주기적 ping.

@app.websocket("/ws")
async def ws_ep(websocket: WebSocket):
    await websocket.accept()
    async def sender():
        while True:
            await asyncio.sleep(30)
            try:
                await websocket.send_json({"type": "ping"})
            except Exception:
                return

    async def receiver():
        while True:
            data = await websocket.receive_json()
            if data.get("type") == "pong":
                continue
            # handle real message

    await asyncio.gather(sender(), receiver())

Nginx / ALB / CloudFront 앞단이 있으면 idle timeout 을 유의 (기본 60초).

테스트

TestClient.websocket_connect 는 context manager:

from fastapi.testclient import TestClient

def test_websocket():
    client = TestClient(app)
    with client.websocket_connect("/ws?token=valid") as ws:
        ws.send_text("hello")
        data = ws.receive_text()
        assert data == "got: hello"

배포 시 유의점

Uvicorn workers

uvicorn main:app --workers 4

WebSocket 은 프로세스 로컬 상태. 다중 워커 broadcast 필요 시 Redis pub/sub.

프록시 (nginx, ALB, CloudFront)

• Upgrade / Connection 헤더 통과 필수 (proxy_set_header Upgrade $http_upgrade)
• Idle timeout 확대 (예: nginx proxy_read_timeout 3600)
• CloudFront: WebSocket 지원, 별도 설정 없음
• ALB: WebSocket 지원, target group 의 stickiness 고려

Kubernetes

• Service type: ClusterIP + Ingress (Nginx ingress 는 annotation 필요)
• Session affinity: 필요 시 Ingress 에서 sticky session

SSE (Server-Sent Events) 와 비교

축	WebSocket	SSE
방향	양방향	서버 -> 클라이언트만
프로토콜	ws/wss	HTTP text/event-stream
재연결	수동	자동 (retry)
바이너리	지원	텍스트만
프록시 호환성	별도 설정 필요	표준 HTTP
적합	채팅, 게임, 협업	로그 스트림, 알림

Client-only push 라면 SSE 가 더 단순하고 안정적.

함정

관련 위키

• FastAPI - 상위 개요
• FastAPI Async - WebSocket 은 반드시 async
• FastAPI DI - WS Depends
• FastAPI Middleware - CORS 미적용
• FastAPI Testing - websocket_connect
• FastAPI Deployment - 프록시 설정