(
self,
path: str,
endpoint: Callable[..., Any],
*,
name: str | None = None,
middleware: Sequence[Middleware] | None = None,
)
| 292 | |
| 293 | class WebSocketRoute(BaseRoute): |
| 294 | def __init__( |
| 295 | self, |
| 296 | path: str, |
| 297 | endpoint: Callable[..., Any], |
| 298 | *, |
| 299 | name: str | None = None, |
| 300 | middleware: Sequence[Middleware] | None = None, |
| 301 | ) -> None: |
| 302 | assert path.startswith("/"), "Routed paths must start with '/'" |
| 303 | self.path = path |
| 304 | self.endpoint = endpoint |
| 305 | self.name = get_name(endpoint) if name is None else name |
| 306 | |
| 307 | endpoint_handler = endpoint |
| 308 | while isinstance(endpoint_handler, functools.partial): |
| 309 | endpoint_handler = endpoint_handler.func |
| 310 | if inspect.isfunction(endpoint_handler) or inspect.ismethod(endpoint_handler): |
| 311 | # Endpoint is function or method. Treat it as `func(websocket)`. |
| 312 | self.app = websocket_session(endpoint) |
| 313 | else: |
| 314 | # Endpoint is a class. Treat it as ASGI. |
| 315 | self.app = endpoint |
| 316 | |
| 317 | if middleware is not None: |
| 318 | for cls, args, kwargs in reversed(middleware): |
| 319 | self.app = cls(self.app, *args, **kwargs) |
| 320 | |
| 321 | self.path_regex, self.path_format, self.param_convertors = compile_path(path) |
| 322 | |
| 323 | def matches(self, scope: Scope) -> tuple[Match, Scope]: |
| 324 | path_params: dict[str, Any] |
nothing calls this directly
no test coverage detected