Heartbeat & Keepalive

import asyncio, time

class HeartbeatManager:
    def __init__(self, *, interval=30, timeout=5):
        self.interval = interval
        self.timeout = timeout
        self.last_pong: Dict[WebSocket, float] = {}

    async def run(self, ws: WebSocket):
        self.last_pong[ws] = time.time()
        try:
            while True:
                await asyncio.sleep(self.interval)
                if ws.client_state.value != 1:  # 1 == CONNECTED
                    return
                await ws.send_json({'type': 'ping'})
                # Give the client `timeout` seconds to pong.
                await asyncio.sleep(self.timeout)
                age = time.time() - self.last_pong.get(ws, 0)
                if age > self.interval + self.timeout:
                    await ws.close(code=4000, reason='heartbeat timeout')
                    return
        finally:
            self.last_pong.pop(ws, None)

    def handle_pong(self, ws: WebSocket):
        self.last_pong[ws] = time.time()

heartbeat = HeartbeatManager()

@app.websocket('/ws')
async def with_heartbeat(websocket: WebSocket):
    await websocket.accept()
    asyncio.create_task(heartbeat.run(websocket))
    try:
        async for msg in websocket.iter_json():
            if msg.get('type') == 'pong':
                heartbeat.handle_pong(websocket)
                continue
            # ... handle other messages
    except WebSocketDisconnect:
        pass