fastapi-ws-batch

Batch and coalesce outgoing FastAPI WebSocket messages, instead of sending one message per event.

The problem

A naive real-time app sends a WebSocket message the instant something happens. If a client emits 30 updates a second (cursor moves, game ticks, live state changes), your server sends 30 separate messages a second — each with its own framing overhead — for just that one client. With more clients, that multiplies fast.

This is a real problem hit while building a real-time multiplayer game: naive per-event sends were flooding the socket. Batching outgoing messages into small windows (~50ms) instead of sending them immediately cut network traffic by roughly 80% in production, with no perceptible added latency.

fastapi-ws-batch packages that fix so you don't have to build it yourself.

Install

pip install fastapi-ws-batch

Usage

Before — one send per event:

@app.websocket("/ws/{room_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: str):
    await websocket.accept()
    while True:
        data = await websocket.receive_json()
        await websocket.send_json(data)  # sent immediately, every time

After — batched automatically:

from fastapi_ws_batch import BatchedConnection

@app.websocket("/ws/{room_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: str):
    await websocket.accept()
    conn = BatchedConnection(websocket, flush_interval=0.05)
    await conn.start()

    try:
        while True:
            data = await websocket.receive_json()
            await conn.add_event(data)  # queued, flushed every 50ms as one batch
    except WebSocketDisconnect:
        pass
    finally:
        await conn.stop()

The receiving client now gets a batch of events every ~50ms instead of a message per event — same information, far fewer round-trips.

Two ways to send

add_event(data) — every message is kept, in order. Use this when every message matters (chat messages, discrete game actions).

add_latest(key, data) — only the newest value per key survives until the next flush; older ones for the same key are silently dropped. Use this for state where only the current value matters (a player's live position, a cursor location):

await conn.add_latest(player_id, {"x": 10, "y": 20})

If a player's position updates 20 times within a 50ms window, only the final position is actually sent — not all 20.

Early flush on bursts

conn = BatchedConnection(websocket, flush_interval=0.05, batch_size=50)

If 50 messages queue up before the next scheduled flush, it flushes immediately instead of waiting out the full interval — so a sudden burst doesn't sit buffered any longer than necessary.

How it works

• add_event() / add_latest() queue messages instead of sending them right away.
• A background task flushes the queue every flush_interval seconds, sending everything queued as one batch via send_json.
• If a failed send happens (e.g. the socket drops mid-send), the batch is restored to the queue instead of being lost, and retried on the next tick — unless something newer for the same key has already replaced it.
• conn.stop() flushes anything left and stops the background task cleanly — always call it when the connection ends (e.g. in a finally block), so nothing queued is lost.

API

BatchedConnection(
    websocket,             # any object with an async send_json(dict) method
    flush_interval,                  # seconds between automatic flushes
    batch_size=None,   # optional: force an early flush once this many messages queue up
)

await conn.start()                    # begin the background flush loop
await conn.add_event(data)            # queue an in-order message
await conn.add_latest(key, data)      # queue a message, keeping only the latest per key
await conn.stop()                     # flush remaining messages and stop cleanly

Why this exists

Existing FastAPI WebSocket libraries handle connection lifecycle and room/broadcast management well, but none of them batch or coalesce outgoing messages — every example sends immediately, per event. This library is the missing piece: put it in front of your existing send logic to cut traffic without changing your app's structure.

License

MIT