Source code for kyoukai.route

Routes are wrapped function objects that are called upon a HTTP request.
import collections
import inspect
import types
import typing

from werkzeug.exceptions import HTTPException
from werkzeug.routing import Rule, Submount
from werkzeug.wrappers import Response

from kyoukai.util import wrap_response

[docs]class Route(object): """ A route object is a wrapped function. They invoke this function when invoked on routing and calling. """ def __init__(self, function, *, reverse_hooks: bool = False, should_invoke_hooks: bool = True, do_argument_checking: bool = True, endpoint: str = None): """ :param function: The underlying callable. This can be a function, or any other callable. :param reverse_hooks: If the request hooks should be reversed for this request (i.e child \ to parent.) :param should_invoke_hooks: If request hooks should be invoked. This is automatically False for error handlers. :param do_argument_checking: If argument type and name checking is enabled for this route. :param endpoint: The custom endpoint for this route. """ if not callable(function): raise TypeError("Route arg must be callable") #: The underlying callable for this route. self._callable = function #: If this route should do argument checking. self.do_argument_checking = do_argument_checking #: The :class:`~.Blueprint` this route is associated with. self.bp = None # type: Blueprint #: A list of tuples (url, methods) for this Route. self.routes = [] #: The custom endpoint for this route. Could be None. self.endpoint = endpoint self.reverse_hooks = reverse_hooks self.should_invoke_hooks = should_invoke_hooks #: Our own specific hooks. self.hooks = {} @property def callable_repr(self): return repr(self._callable)
[docs] def add_path(self, url: str, methods: typing.Sequence[str] = ("GET", "HEAD")): """ Adds a path to the current set of paths for this route. :param url: The routing URL to add. :param methods: An iterable of methods to use for this path. The URL and methods will be added as a pair. """ self.routes.append((url, methods)) return self
[docs] def get_submount(self) -> Submount: """ :return: A submount that represents this route. .. versionadded:: 2.2.0 .. versionchanged:: 2.x.x Changed from getting a list of rules to a single submount object. """ rules = [] for url, methods in self.routes: # mutable list thx methods = list(methods) if "OPTIONS" not in methods: methods.append("OPTIONS") rule = Rule(url, methods=methods, if self.bp is not None else None, endpoint=self.get_endpoint_name()) rules.append(rule) # pass an empty prefix since we don't care about adding that submount = Submount("", rules) return submount
[docs] def get_endpoint_name(self, bp=None) -> str: """ Gets the endpoint name for this route. :param bp: The :class:`.Blueprint` to use for name calculation. :return: The endpoint that can be used. """ if self.endpoint is not None: return self.endpoint if bp is not None: prefix = else: prefix = if self.bp else "" return "{}.{}".format(prefix, self._callable.__name__)
[docs] async def invoke_function(self, ctx, pre_hooks: list, post_hooks: list, params): """ Invokes the underlying callable. This is for use in chaining routes. :param ctx: The :class:`~.HTTPRequestContext` to use for this route. :param pre_hooks: A list of hooks to call before the route is invoked. :param post_hooks: A list of hooks to call after the route is invoked. :param params: The parameters to pass to the function. :return: The result of the invoked function. """ # Invoke the route function. try: # Invoke pre-request hooks, setting `ctx` equal to the new value. if self.should_invoke_hooks: for hook in pre_hooks: _ = await hook(ctx) if _ is not None: ctx = _ if isinstance(params, collections.Mapping): result = self._callable(ctx, **params) else: result = self._callable(ctx, *params) if inspect.isawaitable(result): result = await result result = wrap_response(result, except HTTPException as e: # This is a valid response type raise e else: # Invoke post-request hooks. These happen inside this `else` block because # post-request hooks are only meant to happen if the route invoked successfully. if self.should_invoke_hooks: for hook in post_hooks: _ = await hook(ctx, result) if _ is not None: result = _ return result
[docs] def check_route_args(self, params: dict = None): """ Checks the arguments for a route. :param params: The parameters passed in, as a dict. :raises TypeError: If the arguments passed in were not correct. """ # Get the signature of our callable. sig = inspect.signature(self._callable, follow_wrapped=True) # type: inspect.Signature # signature ignores the `self` param on methods, for some reason # not that i'm complaining f_nargs = len(sig.parameters) - 1 if f_nargs < 0: raise TypeError("Route functions must take ctx argument") # If the lengths of the signature and the params are different, it's obviously wrong. if f_nargs != len(params): raise TypeError("Route takes {} args, passed in {} instead".format(f_nargs, len(params))) # Next, check that all the argument names in the signature are in the params, # so that they can be easily double star expanded into the function. for n, (name, arg) in enumerate(sig.parameters.items()): # Skip the first argument, because it is usually the HTTPRequestContext, and we don't # want to type check that. if n == 0: continue # prevent type checking `self` on methods if isinstance(self._callable, types.MethodType) and n == 1: continue assert isinstance(arg, inspect.Parameter) if not in params: raise ValueError("Argument {} not found in args for callable {}" .format(, self._callable.__name__)) # Also, check that the type of the arg and the annotation matches. value = params[] if arg.annotation is None or arg.annotation is inspect.Parameter.empty: # No annotation, don't type check continue if not isinstance(value, arg.annotation): raise TypeError("Argument {} must be type {} (got type {})".format(, arg.annotation, type(value)) )
[docs] def add_hook(self, type_: str, hook): """ Adds a hook to the current Route. :param type_: The type of hook to add (currently "pre" or "post"). :param hook: The callable function to add as a hook. """ if type_ not in self.hooks: self.hooks[type_] = [] self.hooks[type_].append(hook) return hook
[docs] def get_hooks(self, type_: str): """ Gets the hooks for the current Route for the type. :param type_: The type to get. :return: A list of callables. """ return self.hooks.get(type_, [])
[docs] def before_request(self, func): """ Convenience decorator to add a pre-request hook. """ return self.add_hook(type_="pre", hook=func)
[docs] def after_request(self, func): """ Convenience decorator to add a post-request hook. """ return self.add_hook(type_="post", hook=func)
[docs] async def invoke(self, ctx, args: typing.Iterable[typing.Any] = (), params: typing.Container = None) -> Response: """ Invokes a route. This will run the underlying function. :param ctx: The :class:`~.HTTPRequestContext` which is used in this request. :param args: Any args to expand into the function. :param params: Any keyword params that are used in this request. :return: The result of the route's function. """ if params is None: params = {} # Set some useful attributes on the ctx. ctx.route = self # Invoke the route function. if not params: params = {} if self.do_argument_checking and not args: self.check_route_args(params) else: params = list(args) + list(params.values()) # Try and get the hooks. pre_hooks = self.bp.get_hooks("pre").copy() if self.reverse_hooks: pre_hooks = list(reversed(pre_hooks)) post_hooks = self.bp.get_hooks("post").copy() if self.reverse_hooks: post_hooks = list(reversed(post_hooks)) # merge pre and post hooks with the route-specific ones pre_hooks += self.get_hooks("pre").copy() post_hooks += self.get_hooks("post").copy() return await self.invoke_function(ctx, pre_hooks, post_hooks, params)