822 lines
31 KiB
Python
822 lines
31 KiB
Python
from ast import NodeVisitor, Return, parse
|
|
from contextlib import suppress
|
|
from inspect import getsource, signature
|
|
from textwrap import dedent
|
|
from typing import (
|
|
Any,
|
|
Callable,
|
|
Dict,
|
|
Iterable,
|
|
List,
|
|
Optional,
|
|
Set,
|
|
Tuple,
|
|
Union,
|
|
cast,
|
|
)
|
|
|
|
from sanic_routing.route import Route
|
|
|
|
from sanic.base.meta import SanicMeta
|
|
from sanic.constants import HTTP_METHODS
|
|
from sanic.errorpages import RESPONSE_MAPPING
|
|
from sanic.mixins.base import BaseMixin
|
|
from sanic.models.futures import FutureRoute, FutureStatic
|
|
from sanic.models.handler_types import RouteHandler
|
|
from sanic.types import HashableDict
|
|
|
|
|
|
RouteWrapper = Callable[
|
|
[RouteHandler], Union[RouteHandler, Tuple[Route, RouteHandler]]
|
|
]
|
|
|
|
|
|
class RouteMixin(BaseMixin, metaclass=SanicMeta):
|
|
def __init__(self, *args, **kwargs) -> None:
|
|
self._future_routes: Set[FutureRoute] = set()
|
|
self._future_statics: Set[FutureStatic] = set()
|
|
|
|
def _apply_route(self, route: FutureRoute) -> List[Route]:
|
|
raise NotImplementedError # noqa
|
|
|
|
def route(
|
|
self,
|
|
uri: str,
|
|
methods: Optional[Iterable[str]] = None,
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
stream: bool = False,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
ignore_body: bool = False,
|
|
apply: bool = True,
|
|
subprotocols: Optional[List[str]] = None,
|
|
websocket: bool = False,
|
|
unquote: bool = False,
|
|
static: bool = False,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
**ctx_kwargs: Any,
|
|
) -> RouteWrapper:
|
|
"""Decorate a function to be registered as a route.
|
|
|
|
Args:
|
|
uri (str): Path of the URL.
|
|
methods (Optional[Iterable[str]]): List or tuple of
|
|
methods allowed.
|
|
host (Optional[Union[str, List[str]]]): The host, if required.
|
|
strict_slashes (Optional[bool]): Whether to apply strict slashes
|
|
to the route.
|
|
stream (bool): Whether to allow the request to stream its body.
|
|
version (Optional[Union[int, str, float]]): Route specific
|
|
versioning.
|
|
name (Optional[str]): User-defined route name for url_for.
|
|
ignore_body (bool): Whether the handler should ignore request
|
|
body (e.g. `GET` requests).
|
|
apply (bool): Apply middleware to the route.
|
|
subprotocols (Optional[List[str]]): List of subprotocols.
|
|
websocket (bool): Enable WebSocket support.
|
|
unquote (bool): Unquote special characters in the URL path.
|
|
static (bool): Enable static route.
|
|
version_prefix (str): URL path that should be before the version
|
|
value; default: `"/v"`.
|
|
error_format (Optional[str]): Error format for the route.
|
|
ctx_kwargs (Any): Keyword arguments that begin with a `ctx_*`
|
|
prefix will be appended to the route context (`route.ctx`).
|
|
|
|
Returns:
|
|
RouteWrapper: Tuple of routes, decorated function.
|
|
|
|
Examples:
|
|
Using the method to define a GET endpoint:
|
|
|
|
```python
|
|
@app.route("/hello")
|
|
async def hello(request: Request):
|
|
return text("Hello, World!")
|
|
```
|
|
|
|
Adding context kwargs to the route:
|
|
|
|
```python
|
|
@app.route("/greet", ctx_name="World")
|
|
async def greet(request: Request):
|
|
name = request.route.ctx.name
|
|
return text(f"Hello, {name}!")
|
|
```
|
|
"""
|
|
|
|
# Fix case where the user did not prefix the URL with a /
|
|
# and will probably get confused as to why it's not working
|
|
if not uri.startswith("/") and (uri or hasattr(self, "router")):
|
|
uri = "/" + uri
|
|
|
|
if strict_slashes is None:
|
|
strict_slashes = self.strict_slashes
|
|
|
|
if not methods and not websocket:
|
|
methods = frozenset({"GET"})
|
|
|
|
route_context = self._build_route_context(ctx_kwargs)
|
|
|
|
def decorator(handler):
|
|
nonlocal uri
|
|
nonlocal methods
|
|
nonlocal host
|
|
nonlocal strict_slashes
|
|
nonlocal stream
|
|
nonlocal version
|
|
nonlocal name
|
|
nonlocal ignore_body
|
|
nonlocal subprotocols
|
|
nonlocal websocket
|
|
nonlocal static
|
|
nonlocal version_prefix
|
|
nonlocal error_format
|
|
|
|
if isinstance(handler, tuple):
|
|
# if a handler fn is already wrapped in a route, the handler
|
|
# variable will be a tuple of (existing routes, handler fn)
|
|
_, handler = handler
|
|
|
|
name = self._generate_name(name, handler)
|
|
|
|
if isinstance(host, str):
|
|
host = frozenset([host])
|
|
elif host and not isinstance(host, frozenset):
|
|
try:
|
|
host = frozenset(host)
|
|
except TypeError:
|
|
raise ValueError(
|
|
"Expected either string or Iterable of host strings, "
|
|
"not %s" % host
|
|
)
|
|
if isinstance(subprotocols, list):
|
|
# Ordered subprotocols, maintain order
|
|
subprotocols = tuple(subprotocols)
|
|
elif isinstance(subprotocols, set):
|
|
# subprotocol is unordered, keep it unordered
|
|
subprotocols = frozenset(subprotocols)
|
|
|
|
if not error_format or error_format == "auto":
|
|
error_format = self._determine_error_format(handler)
|
|
|
|
route = FutureRoute(
|
|
handler,
|
|
uri,
|
|
None if websocket else frozenset([x.upper() for x in methods]),
|
|
host,
|
|
strict_slashes,
|
|
stream,
|
|
version,
|
|
name,
|
|
ignore_body,
|
|
websocket,
|
|
subprotocols,
|
|
unquote,
|
|
static,
|
|
version_prefix,
|
|
error_format,
|
|
route_context,
|
|
)
|
|
overwrite = getattr(self, "_allow_route_overwrite", False)
|
|
if overwrite:
|
|
self._future_routes = set(
|
|
filter(lambda x: x.uri != uri, self._future_routes)
|
|
)
|
|
self._future_routes.add(route)
|
|
|
|
args = list(signature(handler).parameters.keys())
|
|
if websocket and len(args) < 2:
|
|
handler_name = handler.__name__
|
|
|
|
raise ValueError(
|
|
f"Required parameter `request` and/or `ws` missing "
|
|
f"in the {handler_name}() route?"
|
|
)
|
|
elif not args:
|
|
handler_name = handler.__name__
|
|
|
|
raise ValueError(
|
|
f"Required parameter `request` missing "
|
|
f"in the {handler_name}() route?"
|
|
)
|
|
|
|
if not websocket and stream:
|
|
handler.is_stream = stream
|
|
|
|
if apply:
|
|
self._apply_route(route, overwrite=overwrite)
|
|
|
|
if static:
|
|
return route, handler
|
|
return handler
|
|
|
|
return decorator
|
|
|
|
def add_route(
|
|
self,
|
|
handler: RouteHandler,
|
|
uri: str,
|
|
methods: Iterable[str] = frozenset({"GET"}),
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
stream: bool = False,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
unquote: bool = False,
|
|
**ctx_kwargs: Any,
|
|
) -> RouteHandler:
|
|
"""A helper method to register class-based view or functions as a handler to the application url routes.
|
|
|
|
Args:
|
|
handler (RouteHandler): Function or class-based view used as a route handler.
|
|
uri (str): Path of the URL.
|
|
methods (Iterable[str]): List or tuple of methods allowed; these are overridden if using an HTTPMethodView.
|
|
host (Optional[Union[str, List[str]]]): Hostname or hostnames to match for this route.
|
|
strict_slashes (Optional[bool]): If set, a route's slashes will be strict. E.g. `/foo` will not match `/foo/`.
|
|
version (Optional[Union[int, str, float]]): Version of the API for this route.
|
|
name (Optional[str]): User-defined route name for `url_for`.
|
|
stream (bool): Boolean specifying if the handler is a stream handler.
|
|
version_prefix (str): URL path that should be before the version value; default: ``/v``.
|
|
error_format (Optional[str]): Custom error format string.
|
|
unquote (bool): Boolean specifying if the handler requires unquoting.
|
|
ctx_kwargs (Any): Keyword arguments that begin with a `ctx_*` prefix will be appended to the route context (``route.ctx``). See below for examples.
|
|
|
|
Returns:
|
|
RouteHandler: The route handler.
|
|
|
|
Examples:
|
|
```python
|
|
from sanic import Sanic, text
|
|
|
|
app = Sanic("test")
|
|
|
|
async def handler(request):
|
|
return text("OK")
|
|
|
|
app.add_route(handler, "/test", methods=["GET", "POST"])
|
|
```
|
|
|
|
You can use `ctx_kwargs` to add custom context to the route. This
|
|
can often be useful when wanting to add metadata to a route that
|
|
can be used by other parts of the application (like middleware).
|
|
|
|
```python
|
|
from sanic import Sanic, text
|
|
|
|
app = Sanic("test")
|
|
|
|
async def handler(request):
|
|
return text("OK")
|
|
|
|
async def custom_middleware(request):
|
|
if request.route.ctx.monitor:
|
|
do_some_monitoring()
|
|
|
|
app.add_route(handler, "/test", methods=["GET", "POST"], ctx_monitor=True)
|
|
app.register_middleware(custom_middleware)
|
|
""" # noqa: E501
|
|
# Handle HTTPMethodView differently
|
|
if hasattr(handler, "view_class"):
|
|
methods = set()
|
|
|
|
for method in HTTP_METHODS:
|
|
view_class = getattr(handler, "view_class")
|
|
_handler = getattr(view_class, method.lower(), None)
|
|
if _handler:
|
|
methods.add(method)
|
|
if hasattr(_handler, "is_stream"):
|
|
stream = True
|
|
|
|
if strict_slashes is None:
|
|
strict_slashes = self.strict_slashes
|
|
|
|
self.route(
|
|
uri=uri,
|
|
methods=methods,
|
|
host=host,
|
|
strict_slashes=strict_slashes,
|
|
stream=stream,
|
|
version=version,
|
|
name=name,
|
|
version_prefix=version_prefix,
|
|
error_format=error_format,
|
|
unquote=unquote,
|
|
**ctx_kwargs,
|
|
)(handler)
|
|
return handler
|
|
|
|
# Shorthand method decorators
|
|
def get(
|
|
self,
|
|
uri: str,
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
ignore_body: bool = True,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
**ctx_kwargs: Any,
|
|
) -> RouteHandler:
|
|
"""Decorate a function handler to create a route definition using the **GET** HTTP method.
|
|
|
|
Args:
|
|
uri (str): URL to be tagged to GET method of HTTP.
|
|
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
|
|
the service to use.
|
|
strict_slashes (Optional[bool]): Instruct Sanic to check if the
|
|
request URLs need to terminate with a `/`.
|
|
version (Optional[Union[int, str, float]]): API Version.
|
|
name (Optional[str]): Unique name that can be used to identify
|
|
the route.
|
|
ignore_body (bool): Whether the handler should ignore request
|
|
body. This means the body of the request, if sent, will not
|
|
be consumed. In that instance, you will see a warning in
|
|
the logs. Defaults to `True`, meaning do not consume the body.
|
|
version_prefix (str): URL path that should be before the version
|
|
value. Defaults to `"/v"`.
|
|
error_format (Optional[str]): Custom error format string.
|
|
**ctx_kwargs (Any): Keyword arguments that begin with a
|
|
`ctx_* prefix` will be appended to the route
|
|
context (`route.ctx`).
|
|
|
|
Returns:
|
|
RouteHandler: Object decorated with route method.
|
|
""" # noqa: E501
|
|
return cast(
|
|
RouteHandler,
|
|
self.route(
|
|
uri,
|
|
methods=frozenset({"GET"}),
|
|
host=host,
|
|
strict_slashes=strict_slashes,
|
|
version=version,
|
|
name=name,
|
|
ignore_body=ignore_body,
|
|
version_prefix=version_prefix,
|
|
error_format=error_format,
|
|
**ctx_kwargs,
|
|
),
|
|
)
|
|
|
|
def post(
|
|
self,
|
|
uri: str,
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
stream: bool = False,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
**ctx_kwargs: Any,
|
|
) -> RouteHandler:
|
|
"""Decorate a function handler to create a route definition using the **POST** HTTP method.
|
|
|
|
Args:
|
|
uri (str): URL to be tagged to POST method of HTTP.
|
|
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
|
|
the service to use.
|
|
strict_slashes (Optional[bool]): Instruct Sanic to check if the
|
|
request URLs need to terminate with a `/`.
|
|
stream (bool): Whether or not to stream the request body.
|
|
Defaults to `False`.
|
|
version (Optional[Union[int, str, float]]): API Version.
|
|
name (Optional[str]): Unique name that can be used to identify
|
|
the route.
|
|
version_prefix (str): URL path that should be before the version
|
|
value. Defaults to `"/v"`.
|
|
error_format (Optional[str]): Custom error format string.
|
|
**ctx_kwargs (Any): Keyword arguments that begin with a
|
|
`ctx_*` prefix will be appended to the route
|
|
context (`route.ctx`).
|
|
|
|
Returns:
|
|
RouteHandler: Object decorated with route method.
|
|
""" # noqa: E501
|
|
return cast(
|
|
RouteHandler,
|
|
self.route(
|
|
uri,
|
|
methods=frozenset({"POST"}),
|
|
host=host,
|
|
strict_slashes=strict_slashes,
|
|
stream=stream,
|
|
version=version,
|
|
name=name,
|
|
version_prefix=version_prefix,
|
|
error_format=error_format,
|
|
**ctx_kwargs,
|
|
),
|
|
)
|
|
|
|
def put(
|
|
self,
|
|
uri: str,
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
stream: bool = False,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
**ctx_kwargs: Any,
|
|
) -> RouteHandler:
|
|
"""Decorate a function handler to create a route definition using the **PUT** HTTP method.
|
|
|
|
Args:
|
|
uri (str): URL to be tagged to PUT method of HTTP.
|
|
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
|
|
the service to use.
|
|
strict_slashes (Optional[bool]): Instruct Sanic to check if the
|
|
request URLs need to terminate with a `/`.
|
|
stream (bool): Whether or not to stream the request body.
|
|
Defaults to `False`.
|
|
version (Optional[Union[int, str, float]]): API Version.
|
|
name (Optional[str]): Unique name that can be used to identify
|
|
the route.
|
|
version_prefix (str): URL path that should be before the version
|
|
value. Defaults to `"/v"`.
|
|
error_format (Optional[str]): Custom error format string.
|
|
**ctx_kwargs (Any): Keyword arguments that begin with a
|
|
`ctx_*` prefix will be appended to the route
|
|
context (`route.ctx`).
|
|
|
|
Returns:
|
|
RouteHandler: Object decorated with route method.
|
|
""" # noqa: E501
|
|
return cast(
|
|
RouteHandler,
|
|
self.route(
|
|
uri,
|
|
methods=frozenset({"PUT"}),
|
|
host=host,
|
|
strict_slashes=strict_slashes,
|
|
stream=stream,
|
|
version=version,
|
|
name=name,
|
|
version_prefix=version_prefix,
|
|
error_format=error_format,
|
|
**ctx_kwargs,
|
|
),
|
|
)
|
|
|
|
def head(
|
|
self,
|
|
uri: str,
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
ignore_body: bool = True,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
**ctx_kwargs: Any,
|
|
) -> RouteHandler:
|
|
"""Decorate a function handler to create a route definition using the **HEAD** HTTP method.
|
|
|
|
Args:
|
|
uri (str): URL to be tagged to HEAD method of HTTP.
|
|
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
|
|
the service to use.
|
|
strict_slashes (Optional[bool]): Instruct Sanic to check if the
|
|
request URLs need to terminate with a `/`.
|
|
version (Optional[Union[int, str, float]]): API Version.
|
|
name (Optional[str]): Unique name that can be used to identify
|
|
the route.
|
|
ignore_body (bool): Whether the handler should ignore request
|
|
body. This means the body of the request, if sent, will not
|
|
be consumed. In that instance, you will see a warning in
|
|
the logs. Defaults to `True`, meaning do not consume the body.
|
|
version_prefix (str): URL path that should be before the version
|
|
value. Defaults to `"/v"`.
|
|
error_format (Optional[str]): Custom error format string.
|
|
**ctx_kwargs (Any): Keyword arguments that begin with a
|
|
`ctx_*` prefix will be appended to the route
|
|
context (`route.ctx`).
|
|
|
|
Returns:
|
|
RouteHandler: Object decorated with route method.
|
|
""" # noqa: E501
|
|
return cast(
|
|
RouteHandler,
|
|
self.route(
|
|
uri,
|
|
methods=frozenset({"HEAD"}),
|
|
host=host,
|
|
strict_slashes=strict_slashes,
|
|
version=version,
|
|
name=name,
|
|
ignore_body=ignore_body,
|
|
version_prefix=version_prefix,
|
|
error_format=error_format,
|
|
**ctx_kwargs,
|
|
),
|
|
)
|
|
|
|
def options(
|
|
self,
|
|
uri: str,
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
ignore_body: bool = True,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
**ctx_kwargs: Any,
|
|
) -> RouteHandler:
|
|
"""Decorate a function handler to create a route definition using the **OPTIONS** HTTP method.
|
|
|
|
Args:
|
|
uri (str): URL to be tagged to OPTIONS method of HTTP.
|
|
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
|
|
the service to use.
|
|
strict_slashes (Optional[bool]): Instruct Sanic to check if the
|
|
request URLs need to terminate with a `/`.
|
|
version (Optional[Union[int, str, float]]): API Version.
|
|
name (Optional[str]): Unique name that can be used to identify
|
|
the route.
|
|
ignore_body (bool): Whether the handler should ignore request
|
|
body. This means the body of the request, if sent, will not
|
|
be consumed. In that instance, you will see a warning in
|
|
the logs. Defaults to `True`, meaning do not consume the body.
|
|
version_prefix (str): URL path that should be before the version
|
|
value. Defaults to `"/v"`.
|
|
error_format (Optional[str]): Custom error format string.
|
|
**ctx_kwargs (Any): Keyword arguments that begin with a
|
|
`ctx_*` prefix will be appended to the route
|
|
context (`route.ctx`).
|
|
|
|
Returns:
|
|
RouteHandler: Object decorated with route method.
|
|
""" # noqa: E501
|
|
return cast(
|
|
RouteHandler,
|
|
self.route(
|
|
uri,
|
|
methods=frozenset({"OPTIONS"}),
|
|
host=host,
|
|
strict_slashes=strict_slashes,
|
|
version=version,
|
|
name=name,
|
|
ignore_body=ignore_body,
|
|
version_prefix=version_prefix,
|
|
error_format=error_format,
|
|
**ctx_kwargs,
|
|
),
|
|
)
|
|
|
|
def patch(
|
|
self,
|
|
uri: str,
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
stream=False,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
**ctx_kwargs: Any,
|
|
) -> RouteHandler:
|
|
"""Decorate a function handler to create a route definition using the **PATCH** HTTP method.
|
|
|
|
Args:
|
|
uri (str): URL to be tagged to PATCH method of HTTP.
|
|
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
|
|
the service to use.
|
|
strict_slashes (Optional[bool]): Instruct Sanic to check if the
|
|
request URLs need to terminate with a `/`.
|
|
stream (bool): Set to `True` if full request streaming is needed,
|
|
`False` otherwise. Defaults to `False`.
|
|
version (Optional[Union[int, str, float]]): API Version.
|
|
name (Optional[str]): Unique name that can be used to identify
|
|
the route.
|
|
version_prefix (str): URL path that should be before the version
|
|
value. Defaults to `"/v"`.
|
|
error_format (Optional[str]): Custom error format string.
|
|
**ctx_kwargs (Any): Keyword arguments that begin with a
|
|
`ctx_*` prefix will be appended to the route
|
|
context (`route.ctx`).
|
|
|
|
Returns:
|
|
RouteHandler: Object decorated with route method.
|
|
""" # noqa: E501
|
|
return cast(
|
|
RouteHandler,
|
|
self.route(
|
|
uri,
|
|
methods=frozenset({"PATCH"}),
|
|
host=host,
|
|
strict_slashes=strict_slashes,
|
|
stream=stream,
|
|
version=version,
|
|
name=name,
|
|
version_prefix=version_prefix,
|
|
error_format=error_format,
|
|
**ctx_kwargs,
|
|
),
|
|
)
|
|
|
|
def delete(
|
|
self,
|
|
uri: str,
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
ignore_body: bool = False,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
**ctx_kwargs: Any,
|
|
) -> RouteHandler:
|
|
"""Decorate a function handler to create a route definition using the **DELETE** HTTP method.
|
|
|
|
Args:
|
|
uri (str): URL to be tagged to the DELETE method of HTTP.
|
|
host (Optional[Union[str, List[str]]]): Host IP or FQDN for the
|
|
service to use.
|
|
strict_slashes (Optional[bool]): Instruct Sanic to check if the
|
|
request URLs need to terminate with a */*.
|
|
version (Optional[Union[int, str, float]]): API Version.
|
|
name (Optional[str]): Unique name that can be used to identify
|
|
the Route.
|
|
ignore_body (bool): Whether or not to ignore the body in the
|
|
request. Defaults to `False`.
|
|
version_prefix (str): URL path that should be before the version
|
|
value. Defaults to `"/v"`.
|
|
error_format (Optional[str]): Custom error format string.
|
|
**ctx_kwargs (Any): Keyword arguments that begin with a `ctx_*`
|
|
prefix will be appended to the route context (`route.ctx`).
|
|
|
|
Returns:
|
|
RouteHandler: Object decorated with route method.
|
|
""" # noqa: E501
|
|
return cast(
|
|
RouteHandler,
|
|
self.route(
|
|
uri,
|
|
methods=frozenset({"DELETE"}),
|
|
host=host,
|
|
strict_slashes=strict_slashes,
|
|
version=version,
|
|
name=name,
|
|
ignore_body=ignore_body,
|
|
version_prefix=version_prefix,
|
|
error_format=error_format,
|
|
**ctx_kwargs,
|
|
),
|
|
)
|
|
|
|
def websocket(
|
|
self,
|
|
uri: str,
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
subprotocols: Optional[List[str]] = None,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
apply: bool = True,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
**ctx_kwargs: Any,
|
|
):
|
|
"""Decorate a function to be registered as a websocket route.
|
|
|
|
Args:
|
|
uri (str): Path of the URL.
|
|
host (Optional[Union[str, List[str]]]): Host IP or FQDN details.
|
|
strict_slashes (Optional[bool]): If the API endpoint needs to
|
|
terminate with a `"/"` or not.
|
|
subprotocols (Optional[List[str]]): Optional list of str with
|
|
supported subprotocols.
|
|
version (Optional[Union[int, str, float]]): WebSocket
|
|
protocol version.
|
|
name (Optional[str]): A unique name assigned to the URL so that
|
|
it can be used with url_for.
|
|
apply (bool): If set to False, it doesn't apply the route to the
|
|
app. Default is `True`.
|
|
version_prefix (str): URL path that should be before the version
|
|
value. Defaults to `"/v"`.
|
|
error_format (Optional[str]): Custom error format string.
|
|
**ctx_kwargs (Any): Keyword arguments that begin with
|
|
a `ctx_* prefix` will be appended to the route
|
|
context (`route.ctx`).
|
|
|
|
Returns:
|
|
tuple: Tuple of routes, decorated function.
|
|
"""
|
|
return self.route(
|
|
uri=uri,
|
|
host=host,
|
|
methods=None,
|
|
strict_slashes=strict_slashes,
|
|
version=version,
|
|
name=name,
|
|
apply=apply,
|
|
subprotocols=subprotocols,
|
|
websocket=True,
|
|
version_prefix=version_prefix,
|
|
error_format=error_format,
|
|
**ctx_kwargs,
|
|
)
|
|
|
|
def add_websocket_route(
|
|
self,
|
|
handler,
|
|
uri: str,
|
|
host: Optional[Union[str, List[str]]] = None,
|
|
strict_slashes: Optional[bool] = None,
|
|
subprotocols=None,
|
|
version: Optional[Union[int, str, float]] = None,
|
|
name: Optional[str] = None,
|
|
version_prefix: str = "/v",
|
|
error_format: Optional[str] = None,
|
|
**ctx_kwargs: Any,
|
|
):
|
|
"""A helper method to register a function as a websocket route.
|
|
|
|
Args:
|
|
handler (Callable): A callable function or instance of a class
|
|
that can handle the websocket request.
|
|
uri (str): URL path that will be mapped to the websocket handler.
|
|
host (Optional[Union[str, List[str]]]): Host IP or FQDN details.
|
|
strict_slashes (Optional[bool]): If the API endpoint needs to
|
|
terminate with a `"/"` or not.
|
|
subprotocols (Optional[List[str]]): Subprotocols to be used with
|
|
websocket handshake.
|
|
version (Optional[Union[int, str, float]]): Versioning information.
|
|
name (Optional[str]): A unique name assigned to the URL.
|
|
version_prefix (str): URL path before the version value.
|
|
Defaults to `"/v"`.
|
|
error_format (Optional[str]): Format for error handling.
|
|
**ctx_kwargs (Any): Keyword arguments beginning with `ctx_*`
|
|
prefix will be appended to the route context (`route.ctx`).
|
|
|
|
Returns:
|
|
Callable: Object passed as the handler.
|
|
"""
|
|
return self.websocket(
|
|
uri=uri,
|
|
host=host,
|
|
strict_slashes=strict_slashes,
|
|
subprotocols=subprotocols,
|
|
version=version,
|
|
name=name,
|
|
version_prefix=version_prefix,
|
|
error_format=error_format,
|
|
**ctx_kwargs,
|
|
)(handler)
|
|
|
|
def _determine_error_format(self, handler) -> str:
|
|
with suppress(OSError, TypeError):
|
|
src = dedent(getsource(handler))
|
|
tree = parse(src)
|
|
http_response_types = self._get_response_types(tree)
|
|
|
|
if len(http_response_types) == 1:
|
|
return next(iter(http_response_types))
|
|
|
|
return ""
|
|
|
|
def _get_response_types(self, node):
|
|
types = set()
|
|
|
|
class HttpResponseVisitor(NodeVisitor):
|
|
def visit_Return(self, node: Return) -> Any:
|
|
nonlocal types
|
|
|
|
with suppress(AttributeError):
|
|
checks = [node.value.func.id] # type: ignore
|
|
if node.value.keywords: # type: ignore
|
|
checks += [
|
|
k.value
|
|
for k in node.value.keywords # type: ignore
|
|
if k.arg == "content_type"
|
|
]
|
|
|
|
for check in checks:
|
|
if check in RESPONSE_MAPPING:
|
|
types.add(RESPONSE_MAPPING[check])
|
|
|
|
HttpResponseVisitor().visit(node)
|
|
|
|
return types
|
|
|
|
def _build_route_context(self, raw: Dict[str, Any]) -> HashableDict:
|
|
ctx_kwargs = {
|
|
key.replace("ctx_", ""): raw.pop(key)
|
|
for key in {**raw}.keys()
|
|
if key.startswith("ctx_")
|
|
}
|
|
if raw:
|
|
unexpected_arguments = ", ".join(raw.keys())
|
|
raise TypeError(
|
|
f"Unexpected keyword arguments: {unexpected_arguments}"
|
|
)
|
|
return HashableDict(ctx_kwargs)
|