from functools import partial from typing import List from sanic.base.meta import SanicMeta from sanic.models.futures import FutureMiddleware class MiddlewareMixin(metaclass=SanicMeta): def __init__(self, *args, **kwargs) -> None: self._future_middleware: List[FutureMiddleware] = [] def _apply_middleware(self, middleware: FutureMiddleware): raise NotImplementedError # noqa def middleware( self, middleware_or_request, attach_to="request", apply=True ): """ Decorate and register middleware to be called before a request is handled or after a response is created. Can either be called as *@app.middleware* or *@app.middleware('request')*. `See user guide re: middleware `__ :param: middleware_or_request: Optional parameter to use for identifying which type of middleware is being registered. """ def register_middleware(middleware, attach_to="request"): nonlocal apply future_middleware = FutureMiddleware(middleware, attach_to) self._future_middleware.append(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 ) def on_request(self, middleware=None): """Register a middleware to be called before a request is handled. This is the same as *@app.middleware('request')*. :param: middleware: A callable that takes in request. """ if callable(middleware): return self.middleware(middleware, "request") else: return partial(self.middleware, attach_to="request") def on_response(self, middleware=None): """Register a middleware to be called after a response is created. This is the same as *@app.middleware('response')*. :param: middleware: A callable that takes in a request and its response. """ if callable(middleware): return self.middleware(middleware, "response") else: return partial(self.middleware, attach_to="response")