Product
Introducing License Enforcement in Socket
Ensure open-source compliance with Socket’s License Enforcement Beta. Set up your License Policy and secure your software!
websockets-routes
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()
@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.
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,
...,
)
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".
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.
FAQs
Routing support for websockets.
We found that websockets-routes demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Product
Ensure open-source compliance with Socket’s License Enforcement Beta. Set up your License Policy and secure your software!
Product
We're launching a new set of license analysis and compliance features for analyzing, managing, and complying with licenses across a range of supported languages and ecosystems.
Product
We're excited to introduce Socket Optimize, a powerful CLI command to secure open source dependencies with tested, optimized package overrides.