Quickstart (Composable Routing)

Updated Dec 08, 2025 
import falcon
from falcon_pachinko import WebSocketResource, WebSocketRouter, handles_message


class ChatResource(WebSocketResource):
    async def on_connect(self, req, ws, room: str) -> bool:
        await ws.accept()
        self.state["room"] = room
        return True  # continue to message handling

    @handles_message("chat.message")
    async def handle_message(self, ws, payload):
        await ws.send_media({"type": "echo", "text": payload.text})


app = falcon.App()
router = WebSocketRouter()
router.add_route("/chat/{room}", ChatResource)
app.add_route("/ws", router)  # router is a Falcon resource
router.mount("/ws")
  • The router is mounted once (router.mount("/ws")) and handles all descendant paths relative to that prefix.
  • Each connection receives a fresh resource instance and a shared state proxy scoped to that connection.