WebSockets

from flaxon import WebSocket

@app.websocket("/ws/echo")
async def echo(socket: WebSocket):
    await socket.accept()

    try:
        async for message in socket.iter_json():
            await socket.send_json({"echo": message})
    except WebSocketDisconnect:
        pass

@app.websocket("/ws/chat/<room_id>")
async def chat(socket: WebSocket, room_id: str):
    await socket.accept()
    await socket.join(room_id)

    try:
        async for message in socket.iter_json():
            await socket.broadcast_json(room_id, {
                "room": room_id,
                "message": message,
                "sender": id(socket),
            })
    finally:
        await socket.leave(room_id)

from flaxon.security import login_required, get_current_user

@app.websocket("/ws/auth")
@login_required
async def auth_websocket(socket: WebSocket):
    user = get_current_user(socket.scope)
    await socket.accept()
    await socket.send_json({"user": user.to_dict()})

from flaxon.websocket import Heartbeat

heartbeat = Heartbeat(interval=30, timeout=60)

@app.websocket("/ws/heartbeat")
async def heartbeat_ws(socket: WebSocket):
    await socket.accept()
    await heartbeat.start(socket)

    try:
        async for message in socket.iter_json():
            # Handle messages
            pass
    finally:
        await heartbeat.stop(socket)

@app.websocket("/ws/multi")
async def multi_room(socket: WebSocket):
    await socket.accept()

    await socket.join("global")
    await socket.join("user-" + str(user_id))

    async for message in socket.iter_json():
        room = message.get("room")
        if room:
            await socket.broadcast_json(room, message)

from flaxon.websocket import WebSocketEvents

@app.websocket("/ws/events")
async def events(socket: WebSocket):
    ws_events = WebSocketEvents(socket)

    @ws_events.on_connect
    async def on_connect():
        print("Client connected")

    @ws_events.on_message
    async def on_message(data):
        await socket.send_json({"echo": data})

    await ws_events.run()