Belajar FastAPI - WebSocket & SSE

#Pendahuluan

Setelah di episode 12 API kalian bisa mengenali pemakainya lewat JWT, sekarang kita buka dimensi baru: realtime. HTTP bekerja dengan pola request-response — klien harus bertanya untuk mendapatkan jawaban. WebSocket dan SSE membaliknya: server bisa mengirim data kapan pun, tanpa ditanya dulu.

Mengapa episode ini penting? Karena fitur-fitur yang membuat aplikasi terasa hidup — chat, notifikasi, update harga, progress bar — semuanya dibangun di atas pola ini. Dan karena FastAPI berbasis ASGI (episode 2), semua ini didukung native tanpa hack.

#WebSocket vs HTTP vs SSE

Aturan praktis: butuh dua arah (klien mengirim dan menerima terus) → WebSocket. Hanya server → klien (misal notifikasi) → SSE jauh lebih sederhana dan otomatis reconnect.

#WebSocket Endpoint Dasar

WebSocket di FastAPI dideklarasikan seperti path operation, tapi dengan @app.websocket:

WebSocket sederhana

from fastapi import FastAPI, WebSocket
 
app = FastAPI()
 
 
@app.websocket("/ws/echo")
async def websocket_echo(websocket: WebSocket) -> None:
    await websocket.accept()
    while True:
        message = await websocket.receive_text()
        await websocket.send_text(f"echo: {message}")

Alur wajib di setiap WebSocket handler:

1. await websocket.accept() — terima handshake.
2. Loop: await websocket.receive_text() menunggu pesan masuk.
3. await websocket.send_text(...) mengirim balik.

Karena handler ini async def, satu event loop bisa melayani banyak koneksi WebSocket sekaligus — di sinilah keunggulan ASGI terasa.

#ConnectionManager: Menangani Banyak Klien

Aplikasi realtime nyata butuh melacak semua koneksi dan mengirim pesan ke semuanya (broadcast). Ini pola ConnectionManager yang dipakai di hampir semua aplikasi FastAPI:

app/ws.py

from fastapi import WebSocket
 
 
class ConnectionManager:
    def __init__(self) -> None:
        self.active_connections: list[WebSocket] = []
 
    async def connect(self, websocket: WebSocket) -> None:
        await websocket.accept()
        self.active_connections.append(websocket)
 
    def disconnect(self, websocket: WebSocket) -> None:
        self.active_connections.remove(websocket)
 
    async def broadcast(self, message: str) -> None:
        for connection in self.active_connections:
            await connection.send_text(message)
 
 
manager = ConnectionManager()

active_connections adalah daftar semua WebSocket yang terhubung. broadcast mengirim pesan ke seluruhnya — fondasi untuk fitur chat dan notifikasi massal.

#Chat dengan ConnectionManager

app/main.py - chat realtime

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from app.ws import manager
 
app = FastAPI()
 
 
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket) -> None:
    await manager.connect(websocket)
    try:
        while True:
            message = await websocket.receive_text()
            await manager.broadcast(f"pesan: {message}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Kunci penting: WebSocketDisconnect di-catch untuk membersihkan koneksi dari daftar. Tanpa ini, koneksi yang putus tetap "berada" di active_connections, dan broadcast akan melempar error saat mengirim ke koneksi mati.

#Uji dengan Klien WebSocket

FastAPI menyediakan TestClient untuk WebSocket juga:

tests/test_ws.py

from fastapi.testclient import TestClient
from app.main import app
 
client = TestClient(app)
 
 
def test_websocket_echo() -> None:
    with client.websocket_connect("/ws/echo") as ws:
        ws.send_text("halo")
        data = ws.receive_text()
        assert data == "echo: halo"

Seluruh logika WebSocket bisa diuji tanpa membuka server — konsisten dengan pola testing episode 10.

#Server-Sent Events (SSE)

Untuk notifikasi satu arah, SSE lebih ringan daripada WebSocket. Klien cukup membuka EventSource di browser, dan server streaming event sebagai text:

SSE dengan StreamingResponse

import asyncio
import json
 
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
 
 
async def event_stream():
    for i in range(5):
        yield f"data: {json.dumps({'counter': i})}\n\n"
        await asyncio.sleep(1)
 
 
@app.get("/events")
async def sse_events() -> StreamingResponse:
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"},
    )

Format SSE mengharuskan baris data: ... diakhiri dua baris kosong. Media type wajib text/event-stream — browser EventSource mengenalinya.

#Autentikasi WebSocket

WebSocket tidak mengirim header Authorization dari browser, jadi token biasanya dikirim lewat query param:

WebSocket dengan auth

from fastapi import WebSocket, WebSocketException, status
 
@app.websocket("/ws/secure")
async def websocket_secure(
    websocket: WebSocket, token: str
) -> None:
    if token != "secret-token":
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    await websocket.accept()

WS_1008_POLICY_VIOLATION adalah code standar untuk penolakan karena kebijakan. Pertukaran token bisa dienkapsulasi sebagai dependency WebSocket — pola yang sama dengan Depends biasa.

#Common Pitfalls

#Penutup

Inti yang harus dibawa pulang:

• WebSocket untuk dua arah; SSE untuk streaming server → klien.
• Pola ConnectionManager melacak semua koneksi dan mendukung broadcast.
• Selalu accept() dulu, lalu loop receive/send, dan catch WebSocketDisconnect.
• SSE = StreamingResponse dengan text/event-stream dan format data: ....
• Autentikasi WebSocket via query param token dengan close code yang tepat.

Di episode 14 selanjutnya kita akan membahas background tasks & async database — menjalankan pekerjaan berat setelah response dikirim, mengombinasikan FastAPI dengan SQLAlchemy async dan asyncpg, serta memahami concurrency di dunia async. API kalian akan semakin cepat dan tangguh!