Skip to main content

Routing support for websockets.

Project description


websockets does not do routing, and I don't like Sanic, so I rolled my own.

Routing backed by Routes.


Decorate your handlers by path, and serve the router.

import asyncio

import websockets
import websockets_routes

router = websockets_routes.Router()

async def thing_list(ws, path):

start_server = websockets.serve(router.handle, ...)

loop = asyncio.get_event_loop()

By default, connections are closed immediately with 4040 if the URL does not match any of the registered routes.

Block connections during handshake

The router has its own serve() method that overrides the process_request() hook, making the server return an HTTP 404 during the handshake phase instead:

start_server = router.serve(...)

This way, a non-matching client never connects to the websocket handler at all.

The override is implemented via a custom WebSocket protocol, so you can subclass that if you need further customisation:

class MyProtocol(websockets_routes.Protocol):

start_server = websockets.serve(

Access route parameters

The handler's second parameter is a RoutedPath instead of a plain str. This is a str subclass, so you can do anything you could as in websockets. There is one additional attribute, params, that allows you to access the matched route parameters.

async def thing_detail(ws, path):
    # Asumming this is accessed with "/thing/123".
    await ws.send(path)  # This sends a text frame "/thing/123".
    await ws.send(path.params["id"])  # Text frame "123".

Per-view handshake hooks

Decorate a class to provide per-view validation and additional processing:

import http

class ThingDetail:
    async def process_request(self, path, headers):
        thing_id = path.params["id"]
        thing = get_thing_or_none(thing_id)
        if thing is not None:
            # Pass additional context to handle().
            path.context["thing"] = thing
            return None
        message = f"thing {thing_id!r} not found\n"
        return (http.HTTPStatus.NOT_FOUND, [], message.encode("utf-8"))

    async def handle(self, ws, path):
        """Now this is only called if thing is found.
        thing = path.context["thing"]  # Retrieve the context to use.

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

websockets-routes-0.2.tar.gz (4.3 kB view hashes)

Uploaded source

Built Distribution

websockets_routes-0.2-py3-none-any.whl (4.0 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page