Move logic into mixins

This commit is contained in:
Adam Hopkins
2021-01-27 10:25:05 +02:00
parent 33d7f4da6b
commit dadf76ce72
8 changed files with 224 additions and 265 deletions

19
sanic/mixins/base.py Normal file
View File

@@ -0,0 +1,19 @@
class Base(type):
def __new__(cls, name, bases, attrs):
init = attrs.get("__init__")
def __init__(self, *args, **kwargs):
nonlocal init
for base in type(self).__bases__:
if base.__name__ != "BaseMixin":
base.__init__(self, *args, **kwargs)
if init:
init(self, *args, **kwargs)
attrs["__init__"] = __init__
return type.__new__(cls, name, bases, attrs)
class BaseMixin(metaclass=Base):
...

View File

@@ -0,0 +1,41 @@
from functools import partial
from typing import Set
from sanic.models.futures import FutureMiddleware
class MiddlewareMixin:
def __init__(self, *args, **kwargs) -> None:
self._future_middleware: Set[FutureMiddleware] = set()
def _apply_middleware(self, middleware: FutureMiddleware):
raise NotImplementedError
def middleware(
self, middleware_or_request, attach_to="request", apply=True
):
"""
Decorate and register middleware to be called before a request.
Can either be called as *@app.middleware* or
*@app.middleware('request')*
:param: middleware_or_request: Optional parameter to use for
identifying which type of middleware is being registered.
"""
def register_middleware(_middleware, attach_to="request"):
future_middleware = FutureMiddleware(_middleware, attach_to)
self._future_middleware.add(future_middleware)
if apply:
self._apply_middleware(future_middleware)
return _middleware
# Detect which way this was called, @middleware or @middleware('AT')
if callable(middleware_or_request):
return register_middleware(
middleware_or_request, attach_to=attach_to
)
else:
return partial(
register_middleware, attach_to=middleware_or_request
)

View File

@@ -1,25 +1,31 @@
from functools import partial
from inspect import signature
from typing import List, Set
from pathlib import PurePath
from typing import List, Set, Union
import websockets
from sanic_routing.route import Route
from sanic.constants import HTTP_METHODS
from sanic.models.futures import FutureRoute
from sanic.models.futures import FutureRoute, FutureStatic
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 __init__(self, *args, **kwargs) -> None:
self._future_routes: Set[FutureRoute] = set()
self._future_statics: Set[FutureStatic] = set()
self.name = ""
self.strict_slashes = False
def _apply_route(self, route: FutureRoute) -> Route:
raise NotImplementedError
def _route(
def _apply_static(self, static: FutureStatic) -> Route:
raise NotImplementedError
def route(
self,
uri,
methods=frozenset({"GET"}),
@@ -133,30 +139,6 @@ class RouteMixin:
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,
@@ -435,7 +417,7 @@ class RouteMixin:
:param version: Blueprint Version
:param name: Unique name to identify the Websocket Route
"""
return self._route(
return self.route(
uri=uri,
host=host,
methods=None,
@@ -474,11 +456,8 @@ class RouteMixin:
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,
uri=uri,
host=host,
strict_slashes=strict_slashes,
subprotocols=subprotocols,
@@ -486,5 +465,69 @@ class RouteMixin:
name=name,
)(handler)
def static(
self,
uri,
file_or_directory: Union[str, bytes, PurePath],
pattern=r"/?.+",
use_modified_since=True,
use_content_range=False,
stream_large_files=False,
name="static",
host=None,
strict_slashes=None,
content_type=None,
apply=True,
):
"""
Register a root to serve files from. The input can either be a
file or a directory. This method will enable an easy and simple way
to setup the :class:`Route` necessary to serve the static files.
:param uri: URL path to be used for serving static content
:param file_or_directory: Path for the Static file/directory with
static files
:param pattern: Regex Pattern identifying the valid static files
:param use_modified_since: If true, send file modified time, and return
not modified if the browser's matches the server's
:param use_content_range: If true, process header for range requests
and sends the file part that is requested
:param stream_large_files: If true, use the
:func:`StreamingHTTPResponse.file_stream` handler rather
than the :func:`HTTPResponse.file` handler to send the file.
If this is an integer, this represents the threshold size to
switch to :func:`StreamingHTTPResponse.file_stream`
:param name: user defined name used for url_for
: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 content_type: user defined content type for header
:return: routes registered on the router
:rtype: List[sanic.router.Route]
"""
if not name.startswith(self.name + "."):
name = f"{self.name}.{name}"
if strict_slashes is None and self.strict_slashes is not None:
strict_slashes = self.strict_slashes
static = FutureStatic(
uri,
file_or_directory,
pattern,
use_modified_since,
use_content_range,
stream_large_files,
name,
host,
strict_slashes,
content_type,
)
self._future_statics.add(static)
if apply:
self._apply_static(static)
def _generate_name(self, handler, name: str) -> str:
return name or handler.__name__