websockets-routes

websockets 不执行路由，我不喜欢Sanic，所以我开发了自定义的。

由Routes支持的路由。

用法

通过路径装饰您的处理程序，并服务路由器。

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()

默认情况下，如果URL不匹配已注册的任何路由，则立即关闭连接并返回4040。

在握手期间阻止连接

路由器有自己的serve()方法，它覆盖了process_request()钩子，使服务器在握手阶段返回HTTP 404

start_server = router.serve(...)

这样，不匹配的客户端永远不会连接到WebSocket处理程序。

该覆盖通过自定义WebSocket协议实现，因此如果您需要进一步的自定义，您可以子类化它

class MyProtocol(websockets_routes.Protocol):
    ...

start_server = websockets.serve(
    router,
    ...,
    create_protocol=MyProtocol,
    ...,
)

访问路由参数

处理程序的第二个参数是RoutedPath而不是普通字符串。这是一个字符串子类，因此您可以像在websockets中那样做任何事情。有一个额外的属性，params，允许您访问匹配的路由参数。

@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".

视图级别的握手钩子

装饰一个类以提供每个视图的验证和附加处理

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.