Schema-Driven Message Dispatch

Updated Dec 08, 2025 
import msgspec
from falcon_pachinko import handles_message


class Message(msgspec.Struct, tag=True):
    type: str


class Join(Message, tag="join"):
    room: str


class ChatResource(WebSocketResource):
    schema = msgspec.defstruct(Join)  # tagged union

    @handles_message("join")
    async def on_join(self, ws, payload: Join):
        await ws.send_media({"type": "joined", "room": payload.room})
  • Messages are decoded with msgspec; unknown tags fall back to on_unhandled(self, ws, raw) when defined.
  • The decorator supports strict=False to allow extra fields when required.