Breakup App and Bluieprint
This commit is contained in:
Adam Hopkins
0
sanic/mixins/__init__.py
Normal file
0
sanic/mixins/__init__.py
Normal file
490
sanic/mixins/routes.py
Normal file
490
sanic/mixins/routes.py
Normal file
@@ -0,0 +1,490 @@
|
||||
from functools import partial
|
||||
from inspect import signature
|
||||
from typing import List, Set
|
||||
|
||||
import websockets
|
||||
|
||||
from sanic_routing.route import Route
|
||||
|
||||
from sanic.constants import HTTP_METHODS
|
||||
from sanic.models.futures import FutureRoute
|
||||
from sanic.views import CompositionView
|
||||
|
||||
|
||||
class RouteMixin:
|
||||
def __init__(self) -> None:
|
||||
self._future_routes: Set[Route] = set()
|
||||
self._future_websocket_routes: Set[Route] = set()
|
||||
|
||||
def _apply_route(self, route: FutureRoute) -> Route:
|
||||
raise NotImplementedError
|
||||
|
||||
def _route(
|
||||
self,
|
||||
uri,
|
||||
methods=frozenset({"GET"}),
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
stream=False,
|
||||
version=None,
|
||||
name=None,
|
||||
ignore_body=False,
|
||||
apply=True,
|
||||
subprotocols=None,
|
||||
websocket=False,
|
||||
):
|
||||
"""Create a blueprint route from a decorated function.
|
||||
|
||||
:param uri: endpoint at which the route will be accessible.
|
||||
:param methods: list of acceptable HTTP methods.
|
||||
:param host: IP Address of FQDN for the sanic server to use.
|
||||
:param strict_slashes: Enforce the API urls are requested with a
|
||||
training */*
|
||||
:param stream: If the route should provide a streaming support
|
||||
:param version: Blueprint Version
|
||||
:param name: Unique name to identify the Route
|
||||
|
||||
:return a decorated method that when invoked will return an object
|
||||
of type :class:`FutureRoute`
|
||||
"""
|
||||
|
||||
if websocket:
|
||||
self.enable_websocket()
|
||||
|
||||
# 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("/"):
|
||||
uri = "/" + uri
|
||||
|
||||
if strict_slashes is None:
|
||||
strict_slashes = self.strict_slashes
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
if websocket:
|
||||
websocket_handler = partial(
|
||||
self._websocket_handler,
|
||||
handler,
|
||||
subprotocols=subprotocols,
|
||||
)
|
||||
websocket_handler.__name__ = (
|
||||
"websocket_handler_" + handler.__name__
|
||||
)
|
||||
websocket_handler.is_websocket = True
|
||||
handler = websocket_handler
|
||||
|
||||
# TODO:
|
||||
# - THink this thru.... do we want all routes namespaced?
|
||||
# -
|
||||
name = self._generate_name(handler, name)
|
||||
|
||||
route = FutureRoute(
|
||||
handler,
|
||||
uri,
|
||||
methods,
|
||||
host,
|
||||
strict_slashes,
|
||||
stream,
|
||||
version,
|
||||
name,
|
||||
ignore_body,
|
||||
)
|
||||
|
||||
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)
|
||||
|
||||
return route, handler
|
||||
|
||||
return decorator
|
||||
|
||||
def route(
|
||||
self,
|
||||
uri,
|
||||
methods=frozenset({"GET"}),
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
stream=False,
|
||||
version=None,
|
||||
name=None,
|
||||
ignore_body=False,
|
||||
apply=True,
|
||||
):
|
||||
return self._route(
|
||||
uri=uri,
|
||||
methods=methods,
|
||||
host=host,
|
||||
strict_slashes=strict_slashes,
|
||||
stream=stream,
|
||||
version=version,
|
||||
name=name,
|
||||
ignore_body=ignore_body,
|
||||
apply=apply,
|
||||
)
|
||||
|
||||
def add_route(
|
||||
self,
|
||||
handler,
|
||||
uri,
|
||||
methods=frozenset({"GET"}),
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
version=None,
|
||||
name=None,
|
||||
stream=False,
|
||||
):
|
||||
"""A helper method to register class instance or
|
||||
functions as a handler to the application url
|
||||
routes.
|
||||
|
||||
:param handler: function or class instance
|
||||
:param uri: path of the URL
|
||||
:param methods: list or tuple of methods allowed, these are overridden
|
||||
if using a HTTPMethodView
|
||||
:param host:
|
||||
:param strict_slashes:
|
||||
:param version:
|
||||
:param name: user defined route name for url_for
|
||||
:param stream: boolean specifying if the handler is a stream handler
|
||||
:return: function or class instance
|
||||
"""
|
||||
# Handle HTTPMethodView differently
|
||||
if hasattr(handler, "view_class"):
|
||||
methods = set()
|
||||
|
||||
for method in HTTP_METHODS:
|
||||
_handler = getattr(handler.view_class, method.lower(), None)
|
||||
if _handler:
|
||||
methods.add(method)
|
||||
if hasattr(_handler, "is_stream"):
|
||||
stream = True
|
||||
|
||||
# handle composition view differently
|
||||
if isinstance(handler, CompositionView):
|
||||
methods = handler.handlers.keys()
|
||||
for _handler in handler.handlers.values():
|
||||
if hasattr(_handler, "is_stream"):
|
||||
stream = True
|
||||
break
|
||||
|
||||
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,
|
||||
)(handler)
|
||||
return handler
|
||||
|
||||
# Shorthand method decorators
|
||||
def get(
|
||||
self,
|
||||
uri,
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
version=None,
|
||||
name=None,
|
||||
ignore_body=True,
|
||||
):
|
||||
"""
|
||||
Add an API URL under the **GET** *HTTP* method
|
||||
|
||||
:param uri: URL to be tagged to **GET** method of *HTTP*
|
||||
:param host: Host IP or FQDN for the service to use
|
||||
:param strict_slashes: Instruct :class:`Sanic` to check if the request
|
||||
URLs need to terminate with a */*
|
||||
:param version: API Version
|
||||
:param name: Unique name that can be used to identify the Route
|
||||
:return: Object decorated with :func:`route` method
|
||||
"""
|
||||
return self.route(
|
||||
uri,
|
||||
methods=frozenset({"GET"}),
|
||||
host=host,
|
||||
strict_slashes=strict_slashes,
|
||||
version=version,
|
||||
name=name,
|
||||
ignore_body=ignore_body,
|
||||
)
|
||||
|
||||
def post(
|
||||
self,
|
||||
uri,
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
stream=False,
|
||||
version=None,
|
||||
name=None,
|
||||
):
|
||||
"""
|
||||
Add an API URL under the **POST** *HTTP* method
|
||||
|
||||
:param uri: URL to be tagged to **POST** method of *HTTP*
|
||||
:param host: Host IP or FQDN for the service to use
|
||||
:param strict_slashes: Instruct :class:`Sanic` to check if the request
|
||||
URLs need to terminate with a */*
|
||||
:param version: API Version
|
||||
:param name: Unique name that can be used to identify the Route
|
||||
:return: Object decorated with :func:`route` method
|
||||
"""
|
||||
return self.route(
|
||||
uri,
|
||||
methods=frozenset({"POST"}),
|
||||
host=host,
|
||||
strict_slashes=strict_slashes,
|
||||
stream=stream,
|
||||
version=version,
|
||||
name=name,
|
||||
)
|
||||
|
||||
def put(
|
||||
self,
|
||||
uri,
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
stream=False,
|
||||
version=None,
|
||||
name=None,
|
||||
):
|
||||
"""
|
||||
Add an API URL under the **PUT** *HTTP* method
|
||||
|
||||
:param uri: URL to be tagged to **PUT** method of *HTTP*
|
||||
:param host: Host IP or FQDN for the service to use
|
||||
:param strict_slashes: Instruct :class:`Sanic` to check if the request
|
||||
URLs need to terminate with a */*
|
||||
:param version: API Version
|
||||
:param name: Unique name that can be used to identify the Route
|
||||
:return: Object decorated with :func:`route` method
|
||||
"""
|
||||
return self.route(
|
||||
uri,
|
||||
methods=frozenset({"PUT"}),
|
||||
host=host,
|
||||
strict_slashes=strict_slashes,
|
||||
stream=stream,
|
||||
version=version,
|
||||
name=name,
|
||||
)
|
||||
|
||||
def head(
|
||||
self,
|
||||
uri,
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
version=None,
|
||||
name=None,
|
||||
ignore_body=True,
|
||||
):
|
||||
return self.route(
|
||||
uri,
|
||||
methods=frozenset({"HEAD"}),
|
||||
host=host,
|
||||
strict_slashes=strict_slashes,
|
||||
version=version,
|
||||
name=name,
|
||||
ignore_body=ignore_body,
|
||||
)
|
||||
|
||||
def options(
|
||||
self,
|
||||
uri,
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
version=None,
|
||||
name=None,
|
||||
ignore_body=True,
|
||||
):
|
||||
"""
|
||||
Add an API URL under the **OPTIONS** *HTTP* method
|
||||
|
||||
:param uri: URL to be tagged to **OPTIONS** method of *HTTP*
|
||||
:param host: Host IP or FQDN for the service to use
|
||||
:param strict_slashes: Instruct :class:`Sanic` to check if the request
|
||||
URLs need to terminate with a */*
|
||||
:param version: API Version
|
||||
:param name: Unique name that can be used to identify the Route
|
||||
:return: Object decorated with :func:`route` method
|
||||
"""
|
||||
return self.route(
|
||||
uri,
|
||||
methods=frozenset({"OPTIONS"}),
|
||||
host=host,
|
||||
strict_slashes=strict_slashes,
|
||||
version=version,
|
||||
name=name,
|
||||
ignore_body=ignore_body,
|
||||
)
|
||||
|
||||
def patch(
|
||||
self,
|
||||
uri,
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
stream=False,
|
||||
version=None,
|
||||
name=None,
|
||||
):
|
||||
"""
|
||||
Add an API URL under the **PATCH** *HTTP* method
|
||||
|
||||
:param uri: URL to be tagged to **PATCH** method of *HTTP*
|
||||
:param host: Host IP or FQDN for the service to use
|
||||
:param strict_slashes: Instruct :class:`Sanic` to check if the request
|
||||
URLs need to terminate with a */*
|
||||
:param version: API Version
|
||||
:param name: Unique name that can be used to identify the Route
|
||||
:return: Object decorated with :func:`route` method
|
||||
"""
|
||||
return self.route(
|
||||
uri,
|
||||
methods=frozenset({"PATCH"}),
|
||||
host=host,
|
||||
strict_slashes=strict_slashes,
|
||||
stream=stream,
|
||||
version=version,
|
||||
name=name,
|
||||
)
|
||||
|
||||
def delete(
|
||||
self,
|
||||
uri,
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
version=None,
|
||||
name=None,
|
||||
ignore_body=True,
|
||||
):
|
||||
"""
|
||||
Add an API URL under the **DELETE** *HTTP* method
|
||||
|
||||
:param uri: URL to be tagged to **DELETE** method of *HTTP*
|
||||
:param host: Host IP or FQDN for the service to use
|
||||
:param strict_slashes: Instruct :class:`Sanic` to check if the request
|
||||
URLs need to terminate with a */*
|
||||
:param version: API Version
|
||||
:param name: Unique name that can be used to identify the Route
|
||||
:return: Object decorated with :func:`route` method
|
||||
"""
|
||||
return self.route(
|
||||
uri,
|
||||
methods=frozenset({"DELETE"}),
|
||||
host=host,
|
||||
strict_slashes=strict_slashes,
|
||||
version=version,
|
||||
name=name,
|
||||
ignore_body=ignore_body,
|
||||
)
|
||||
|
||||
def websocket(
|
||||
self,
|
||||
uri,
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
version=None,
|
||||
name=None,
|
||||
subprotocols=None,
|
||||
apply: bool = True,
|
||||
):
|
||||
"""Create a blueprint websocket route from a decorated function.
|
||||
|
||||
:param uri: endpoint at which the route will be accessible.
|
||||
:param host: IP Address of FQDN for the sanic server to use.
|
||||
:param strict_slashes: Enforce the API urls are requested with a
|
||||
training */*
|
||||
:param version: Blueprint Version
|
||||
:param name: Unique name to identify the Websocket Route
|
||||
"""
|
||||
return self._route(
|
||||
uri=uri,
|
||||
host=host,
|
||||
methods=None,
|
||||
strict_slashes=strict_slashes,
|
||||
version=version,
|
||||
name=name,
|
||||
apply=apply,
|
||||
subprotocols=subprotocols,
|
||||
websocket=True,
|
||||
)
|
||||
|
||||
def add_websocket_route(
|
||||
self,
|
||||
handler,
|
||||
uri,
|
||||
host=None,
|
||||
strict_slashes=None,
|
||||
subprotocols=None,
|
||||
version=None,
|
||||
name=None,
|
||||
):
|
||||
"""
|
||||
A helper method to register a function as a websocket route.
|
||||
|
||||
:param handler: a callable function or instance of a class
|
||||
that can handle the websocket request
|
||||
:param host: Host IP or FQDN details
|
||||
:param uri: URL path that will be mapped to the websocket
|
||||
handler
|
||||
handler
|
||||
:param strict_slashes: If the API endpoint needs to terminate
|
||||
with a "/" or not
|
||||
:param subprotocols: Subprotocols to be used with websocket
|
||||
handshake
|
||||
:param name: A unique name assigned to the URL so that it can
|
||||
be used with :func:`url_for`
|
||||
:return: Objected decorated by :func:`websocket`
|
||||
"""
|
||||
if strict_slashes is None:
|
||||
strict_slashes = self.strict_slashes
|
||||
|
||||
return self.websocket(
|
||||
uri,
|
||||
host=host,
|
||||
strict_slashes=strict_slashes,
|
||||
subprotocols=subprotocols,
|
||||
version=version,
|
||||
name=name,
|
||||
)(handler)
|
||||
|
||||
def _generate_name(self, handler, name: str) -> str:
|
||||
return name or handler.__name__
|
||||
Reference in New Issue
Block a user