bazarr/libs/flask_restx/namespace.py

import inspect
import warnings
import logging
from collections import namedtuple, OrderedDict

from flask import request
from flask.views import http_method_funcs

from ._http import HTTPStatus
from .errors import abort
from .marshalling import marshal, marshal_with
from .model import Model, OrderedModel, SchemaModel
from .reqparse import RequestParser
from .utils import merge

# Container for each route applied to a Resource using @ns.route decorator
ResourceRoute = namedtuple("ResourceRoute", "resource urls route_doc kwargs")


class Namespace(object):
    """
    Group resources together.

    Namespace is to API what :class:`flask:flask.Blueprint` is for :class:`flask:flask.Flask`.

    :param str name: The namespace name
    :param str description: An optional short description
    :param str path: An optional prefix path. If not provided, prefix is ``/+name``
    :param list decorators: A list of decorators to apply to each resources
    :param bool validate: Whether or not to perform validation on this namespace
    :param bool ordered: Whether or not to preserve order on models and marshalling
    :param Api api: an optional API to attache to the namespace
    """

    def __init__(
        self,
        name,
        description=None,
        path=None,
        decorators=None,
        validate=None,
        authorizations=None,
        ordered=False,
        **kwargs
    ):
        self.name = name
        self.description = description
        self._path = path

        self._schema = None
        self._validate = validate
        self.models = {}
        self.urls = {}
        self.decorators = decorators if decorators else []
        self.resources = []  # List[ResourceRoute]
        self.error_handlers = OrderedDict()
        self.default_error_handler = None
        self.authorizations = authorizations
        self.ordered = ordered
        self.apis = []
        if "api" in kwargs:
            self.apis.append(kwargs["api"])
        self.logger = logging.getLogger(__name__ + "." + self.name)

    @property
    def path(self):
        return (self._path or ("/" + self.name)).rstrip("/")

    def add_resource(self, resource, *urls, **kwargs):
        """
        Register a Resource for a given API Namespace

        :param Resource resource: the resource ro register
        :param str urls: one or more url routes to match for the resource,
                         standard flask routing rules apply.
                         Any url variables will be passed to the resource method as args.
        :param str endpoint: endpoint name (defaults to :meth:`Resource.__name__.lower`
            Can be used to reference this route in :class:`fields.Url` fields
        :param list|tuple resource_class_args: args to be forwarded to the constructor of the resource.
        :param dict resource_class_kwargs: kwargs to be forwarded to the constructor of the resource.

        Additional keyword arguments not specified above will be passed as-is
        to :meth:`flask.Flask.add_url_rule`.

        Examples::

            namespace.add_resource(HelloWorld, '/', '/hello')
            namespace.add_resource(Foo, '/foo', endpoint="foo")
            namespace.add_resource(FooSpecial, '/special/foo', endpoint="foo")
        """
        route_doc = kwargs.pop("route_doc", {})
        self.resources.append(ResourceRoute(resource, urls, route_doc, kwargs))
        for api in self.apis:
            ns_urls = api.ns_urls(self, urls)
            api.register_resource(self, resource, *ns_urls, **kwargs)

    def route(self, *urls, **kwargs):
        """
        A decorator to route resources.
        """

        def wrapper(cls):
            doc = kwargs.pop("doc", None)
            if doc is not None:
                # build api doc intended only for this route
                kwargs["route_doc"] = self._build_doc(cls, doc)
            self.add_resource(cls, *urls, **kwargs)
            return cls

        return wrapper

    def _build_doc(self, cls, doc):
        if doc is False:
            return False
        unshortcut_params_description(doc)
        handle_deprecations(doc)
        for http_method in http_method_funcs:
            if http_method in doc:
                if doc[http_method] is False:
                    continue
                unshortcut_params_description(doc[http_method])
                handle_deprecations(doc[http_method])
                if "expect" in doc[http_method] and not isinstance(
                    doc[http_method]["expect"], (list, tuple)
                ):
                    doc[http_method]["expect"] = [doc[http_method]["expect"]]
        return merge(getattr(cls, "__apidoc__", {}), doc)

    def doc(self, shortcut=None, **kwargs):
        """A decorator to add some api documentation to the decorated object"""
        if isinstance(shortcut, str):
            kwargs["id"] = shortcut
        show = shortcut if isinstance(shortcut, bool) else True

        def wrapper(documented):
            documented.__apidoc__ = self._build_doc(
                documented, kwargs if show else False
            )
            return documented

        return wrapper

    def hide(self, func):
        """A decorator to hide a resource or a method from specifications"""
        return self.doc(False)(func)

    def abort(self, *args, **kwargs):
        """
        Properly abort the current request

        See: :func:`~flask_restx.errors.abort`
        """
        abort(*args, **kwargs)

    def add_model(self, name, definition):
        self.models[name] = definition
        for api in self.apis:
            api.models[name] = definition
        return definition

    def model(self, name=None, model=None, mask=None, strict=False, **kwargs):
        """
        Register a model

        :param bool strict - should model validation raise error when non-specified param
                             is provided?

        .. seealso:: :class:`Model`
        """
        cls = OrderedModel if self.ordered else Model
        model = cls(name, model, mask=mask, strict=strict)
        model.__apidoc__.update(kwargs)
        return self.add_model(name, model)

    def schema_model(self, name=None, schema=None):
        """
        Register a model

        .. seealso:: :class:`Model`
        """
        model = SchemaModel(name, schema)
        return self.add_model(name, model)

    def extend(self, name, parent, fields):
        """
        Extend a model (Duplicate all fields)

        :deprecated: since 0.9. Use :meth:`clone` instead
        """
        if isinstance(parent, list):
            parents = parent + [fields]
            model = Model.extend(name, *parents)
        else:
            model = Model.extend(name, parent, fields)
        return self.add_model(name, model)

    def clone(self, name, *specs):
        """
        Clone a model (Duplicate all fields)

        :param str name: the resulting model name
        :param specs: a list of models from which to clone the fields

        .. seealso:: :meth:`Model.clone`

        """
        model = Model.clone(name, *specs)
        return self.add_model(name, model)

    def inherit(self, name, *specs):
        """
        Inherit a model (use the Swagger composition pattern aka. allOf)

        .. seealso:: :meth:`Model.inherit`
        """
        model = Model.inherit(name, *specs)
        return self.add_model(name, model)

    def expect(self, *inputs, **kwargs):
        """
        A decorator to Specify the expected input model

        :param ModelBase|Parse inputs: An expect model or request parser
        :param bool validate: whether to perform validation or not

        """
        expect = []
        params = {"validate": kwargs.get("validate", self._validate), "expect": expect}
        for param in inputs:
            expect.append(param)
        return self.doc(**params)

    def parser(self):
        """Instanciate a :class:`~RequestParser`"""
        return RequestParser()

    def as_list(self, field):
        """Allow to specify nested lists for documentation"""
        field.__apidoc__ = merge(getattr(field, "__apidoc__", {}), {"as_list": True})
        return field

    def marshal_with(
        self, fields, as_list=False, code=HTTPStatus.OK, description=None, **kwargs
    ):
        """
        A decorator specifying the fields to use for serialization.

        :param bool as_list: Indicate that the return type is a list (for the documentation)
        :param int code: Optionally give the expected HTTP response code if its different from 200

        """

        def wrapper(func):
            doc = {
                "responses": {
                    str(code): (description, [fields], kwargs)
                    if as_list
                    else (description, fields, kwargs)
                },
                "__mask__": kwargs.get(
                    "mask", True
                ),  # Mask values can't be determined outside app context
            }
            func.__apidoc__ = merge(getattr(func, "__apidoc__", {}), doc)
            return marshal_with(fields, ordered=self.ordered, **kwargs)(func)

        return wrapper

    def marshal_list_with(self, fields, **kwargs):
        """A shortcut decorator for :meth:`~Api.marshal_with` with ``as_list=True``"""
        return self.marshal_with(fields, True, **kwargs)

    def marshal(self, *args, **kwargs):
        """A shortcut to the :func:`marshal` helper"""
        return marshal(*args, **kwargs)

    def errorhandler(self, exception):
        """A decorator to register an error handler for a given exception"""
        if inspect.isclass(exception) and issubclass(exception, Exception):
            # Register an error handler for a given exception
            def wrapper(func):
                self.error_handlers[exception] = func
                return func

            return wrapper
        else:
            # Register the default error handler
            self.default_error_handler = exception
            return exception

    def param(self, name, description=None, _in="query", **kwargs):
        """
        A decorator to specify one of the expected parameters

        :param str name: the parameter name
        :param str description: a small description
        :param str _in: the parameter location `(query|header|formData|body|cookie)`
        """
        param = kwargs
        param["in"] = _in
        param["description"] = description
        return self.doc(params={name: param})

    def response(self, code, description, model=None, **kwargs):
        """
        A decorator to specify one of the expected responses

        :param int code: the HTTP status code
        :param str description: a small description about the response
        :param ModelBase model: an optional response model

        """
        return self.doc(responses={str(code): (description, model, kwargs)})

    def header(self, name, description=None, **kwargs):
        """
        A decorator to specify one of the expected headers

        :param str name: the HTTP header name
        :param str description: a description about the header

        """
        header = {"description": description}
        header.update(kwargs)
        return self.doc(headers={name: header})

    def produces(self, mimetypes):
        """A decorator to specify the MIME types the API can produce"""
        return self.doc(produces=mimetypes)

    def deprecated(self, func):
        """A decorator to mark a resource or a method as deprecated"""
        return self.doc(deprecated=True)(func)

    def vendor(self, *args, **kwargs):
        """
        A decorator to expose vendor extensions.

        Extensions can be submitted as dict or kwargs.
        The ``x-`` prefix is optionnal and will be added if missing.

        See: http://swagger.io/specification/#specification-extensions-128
        """
        for arg in args:
            kwargs.update(arg)
        return self.doc(vendor=kwargs)

    @property
    def payload(self):
        """Store the input payload in the current request context"""
        return request.get_json()


def unshortcut_params_description(data):
    if "params" in data:
        for name, description in data["params"].items():
            if isinstance(description, str):
                data["params"][name] = {"description": description}


def handle_deprecations(doc):
    if "parser" in doc:
        warnings.warn(
            "The parser attribute is deprecated, use expect instead",
            DeprecationWarning,
            stacklevel=2,
        )
        doc["expect"] = doc.get("expect", []) + [doc.pop("parser")]
    if "body" in doc:
        warnings.warn(
            "The body attribute is deprecated, use expect instead",
            DeprecationWarning,
            stacklevel=2,
        )
        doc["expect"] = doc.get("expect", []) + [doc.pop("body")]