Real-time: WebSockets and server-sent events

import asyncio
import json
from collections import defaultdict

from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI(title="Brew & Bean live order status")

# Demo only: lets index.html opened from a file talk to this API (needed for SSE).
# In a real app, list your website's exact origin instead of "*".
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

# Pretend database: three orders, all just received
orders = {i: {"id": i, "status": "received"} for i in (1, 2, 3)}


class StatusUpdate(BaseModel):
    status: str


# ---------- WebSocket: who is connected, per order ----------
ws_clients: dict[int, set[WebSocket]] = defaultdict(set)

# ---------- SSE: one queue per connected browser, per order ----------
sse_queues: dict[int, set[asyncio.Queue]] = defaultdict(set)


async def publish(order_id: int, message: dict) -> None:
    """Send a message to everyone watching this order, over both channels."""
    for ws in list(ws_clients[order_id]):
        try:
            await ws.send_json(message)
        except Exception:
            ws_clients[order_id].discard(ws)  # connection died
    for queue in list(sse_queues[order_id]):
        queue.put_nowait(message)


@app.patch("/orders/{order_id}/status")
async def update_status(order_id: int, body: StatusUpdate):
    order = orders.get(order_id)
    if order is None:
        raise HTTPException(status_code=404, detail="No such order")
    order["status"] = body.status
    await publish(order_id, {"order_id": order_id, "status": body.status})
    return order


@app.websocket("/ws/orders/{order_id}")
async def order_socket(websocket: WebSocket, order_id: int):
    if order_id not in orders:
        await websocket.close(code=1008)  # policy violation: unknown order
        return
    await websocket.accept()
    ws_clients[order_id].add(websocket)
    try:
        # Send the current state straight away, then keep the connection open
        await websocket.send_json({"order_id": order_id, "status": orders[order_id]["status"]})
        while True:
            await websocket.receive_text()  # the client may send "ping"; we just wait
    except WebSocketDisconnect:
        pass
    finally:
        ws_clients[order_id].discard(websocket)


@app.get("/orders/{order_id}/events")
async def order_events(order_id: int, request: Request):
    if order_id not in orders:
        raise HTTPException(status_code=404, detail="No such order")
    queue: asyncio.Queue = asyncio.Queue()
    sse_queues[order_id].add(queue)

    async def stream():
        try:
            current = {"order_id": order_id, "status": orders[order_id]["status"]}
            yield f"data: {json.dumps(current)}\n\n"
            while True:
                if await request.is_disconnected():
                    break
                try:
                    message = await asyncio.wait_for(queue.get(), timeout=15)
                except asyncio.TimeoutError:
                    yield ": keep-alive\n\n"  # a comment line stops proxies closing an idle stream
                    continue
                yield f"data: {json.dumps(message)}\n\n"
        finally:
            sse_queues[order_id].discard(queue)

    return StreamingResponse(
        stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

# Terminal A: stay connected and listen (-N turns off curl's buffering)
curl -N http://127.0.0.1:8000/orders/2/events

# Terminal B: change the order
curl -X PATCH http://127.0.0.1:8000/orders/2/status -H "Content-Type: application/json" -d '{"status":"preparing"}'

<!DOCTYPE html>
<html>
<body>
  <h2>Order 1</h2>
  <p>WebSocket: <b id="ws">connecting...</b></p>
  <p>SSE: <b id="sse">connecting...</b></p>
  <script>
    // WebSocket version
    const socket = new WebSocket("ws://127.0.0.1:8000/ws/orders/1");
    socket.onmessage = (event) => {
      document.getElementById("ws").textContent = JSON.parse(event.data).status;
    };
    socket.onclose = () => {
      document.getElementById("ws").textContent += " (disconnected)";
    };

    // Server-sent events version
    const source = new EventSource("http://127.0.0.1:8000/orders/1/events");
    source.onmessage = (event) => {
      document.getElementById("sse").textContent = JSON.parse(event.data).status;
    };
  </script>
</body>
</html>

curl -X PATCH http://127.0.0.1:8000/orders/1/status -H "Content-Type: application/json" -d '{"status":"ready"}'

I am building [describe your app]. I need users to see [describe live update, e.g. new comments, delivery location, quiz scores] without refreshing. Compare polling, SSE and WebSocket for it. Recommend one, and describe the message format, how to authenticate the connection, what happens on reconnect, and how it works when I run three server instances.