Skip to main content

Routing support for websockets.

Project description

websockets-routes

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

Routing backed by Routes.

Usage

Decorate your handlers by path, and serve the router.

import asyncio

import websockets
import websockets_routes

router = websockets_routes.Router()

@router.route("/thing/")
async def thing_list(ws, path):
    ...

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

loop = asyncio.get_event_loop()
loop.run_until_complete(start_server)
loop.run_forever()

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(
    router,
    ...,
    create_protocol=MyProtocol,
    ...,
)

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.

@router.route("/thing/{id}")
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

@router.route("/thing/{id}")
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.

Files for websockets-routes, version 0.2
Filename, size File type Python version Upload date Hashes
Filename, size websockets_routes-0.2-py3-none-any.whl (4.0 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size websockets-routes-0.2.tar.gz (4.3 kB) File type Source Python version None Upload date Hashes View

Supported by

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