restapi.py

# Copyright 2018 ACSONE SA/NV
# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).

import functools

from cerberus import Validator

from odoo import _, http
from odoo.exceptions import UserError

from .tools import cerberus_to_json


def method(routes, input_param=None, output_param=None, **kw):
    """Decorator marking the decorated method as being a handler for
      REST requests. The method must be part of a component inheriting from
    ``base.rest.service``.

      :param routes: list of tuple (path, http method). path is a string or
                    array.
                    Each tuple determines which http requests and http method
                    will match the decorated method. The path part can be a
                    single string or an array of strings. See werkzeug's routing
                    documentation for the format of path expression (
                    http://werkzeug.pocoo.org/docs/routing/ ).
      :param: input_param: An instance of an object that implemented
                    ``RestMethodParam``. When processing a request, the http
                    handler first call the from_request method and then call the
                    decorated method with the result of this call.
      :param: output_param: An instance of an object that implemented
                    ``RestMethodParam``. When processing the result of the
                    call to the decorated method, the http handler first call
                    the `to_response` method with this result and then return
                    the result of this call.
      :param auth: The type of authentication method. A special auth method
                   named 'public_or_default' can be used. In such a case
                   when the HTTP route will be generated, the auth method
                   will be computed from the '_default_auth' property defined
                   on the controller with 'public_or_' as prefix.
                   The purpose of 'public_or_default' auth method is to provide
                   a way to specify that a method should work for anonymous users
                   but can be enhanced when an authenticated user is know.
                   It implies that when the 'default' auth part of 'public_or_default'
                   will be replaced by the default_auth specified on the controller
                   in charge of registering the web services, an auth method with
                   the same name is defined into odoo to provide such a behavior.
                   In the following example, the auth method on my ping service
                   will be `public_or_jwt` since this authentication method is
                   provided by the auth_jwt addon.

                    .. code-block:: python

                        class PingService(Component):
                            _inherit = "base.rest.service"
                            _name = "ping_service"
                            _usage = "ping"
                            _collection = "test.api.services"

                            @restapi.method(
                                [(["/<string:message>""], "GET")],
                                auth="public_or_auth",
                            )
                            def _ping(self, message):
                                return {"message": message}


                        class MyRestController(main.RestController):
                            _root_path = '/test/'
                            _collection_name = "test.api.services"
                            _default_auth = "jwt'

      :param cors: The Access-Control-Allow-Origin cors directive value. When
                   set, this automatically adds OPTIONS to allowed http methods
                   so the Odoo request handler will accept it.
      :param bool csrf: Whether CSRF protection should be enabled for the route.
                        Defaults to ``False``
      :param bool save_session: Whether HTTP session should be saved into the
                                session store: Default to ``True``

    """

    def decorator(f):
        _routes = []
        for paths, http_methods in routes:
            if not isinstance(paths, list):
                paths = [paths]
            if not isinstance(http_methods, list):
                http_methods = [http_methods]
            if kw.get("cors") and "OPTIONS" not in http_methods:
                http_methods.append("OPTIONS")
            for m in http_methods:
                _routes.append(([p for p in paths], m))
        routing = {
            "routes": _routes,
            "input_param": input_param,
            "output_param": output_param,
        }
        routing.update(kw)

        @functools.wraps(f)
        def response_wrap(*args, **kw):
            response = f(*args, **kw)
            return response

        response_wrap.routing = routing
        response_wrap.original_func = f
        return response_wrap

    return decorator


class RestMethodParam(object):
    def from_params(self, service, params):
        """
        This method is called to process the parameters received at the
        controller. This method should validate and sanitize these paramaters.
        It could also be used to transform these parameters into the format
        expected by the called method
        :param service:
        :param request: `HttpRequest.params`
        :return: Value into the format expected by the method
        """

    def to_response(self, service, result):
        """
        This method is called to prepare the result of the call to the method
        in a format suitable by the controller (http.Response or JSON dict).
        It's responsible for validating and sanitizing the result.
        :param service:
        :param obj:
        :return: http.Response or JSON dict
        """

    def to_openapi_query_parameters(self, service):
        return {}

    def to_openapi_requestbody(self, service):
        return {}

    def to_openapi_responses(self, service):
        return {}


class BinaryData(RestMethodParam):
    def __init__(self, mediatypes="*/*", required=False):
        if not isinstance(mediatypes, list):
            mediatypes = [mediatypes]
        self._mediatypes = mediatypes
        self._required = required

    @property
    def _binary_content_schema(self):
        return {
            mediatype: {
                "schema": {
                    "type": "string",
                    "format": "binary",
                    "required": self._required,
                }
            }
            for mediatype in self._mediatypes
        }

    def to_openapi_requestbody(self, service):
        return {"content": self._binary_content_schema}

    def to_openapi_responses(self, service):
        return {"200": {"content": self._binary_content_schema}}

    def to_response(self, service, result):
        if not isinstance(result, http.Response):
            # The response has not been build by the called method...
            result = self._to_http_response(result)
        return result

    def from_params(self, service, params):
        return params

    def _to_http_response(self, result):
        mediatype = self._mediatypes[0] if len(self._mediatypes) == 1 else "*/*"
        headers = [
            ("Content-Type", mediatype),
            ("X-Content-Type-Options", "nosniff"),
            ("Content-Disposition", http.content_disposition("file")),
            ("Content-Length", len(result)),
        ]
        return http.request.make_response(result, headers)


class CerberusValidator(RestMethodParam):
    def __init__(self, schema):
        """

        :param schema: can be dict as cerberus schema, an instance of
                       cerberus.Validator or a sting with the method name to
                       call on the service to get the schema or the validator
        """
        self._schema = schema

    def from_params(self, service, params):
        validator = self.get_cerberus_validator(service, "input")
        if validator.validate(params):
            return validator.document
        raise UserError(_("BadRequest %s") % validator.errors)

    def to_response(self, service, result):
        validator = self.get_cerberus_validator(service, "output")
        if validator.validate(result):
            return validator.document
        raise SystemError(_("Invalid Response %s") % validator.errors)

    def to_openapi_query_parameters(self, service):
        json_schema = self.to_json_schema(service, "input")
        parameters = []
        for prop, spec in list(json_schema["properties"].items()):
            params = {
                "name": prop,
                "in": "query",
                "required": prop in json_schema["required"],
                "allowEmptyValue": spec.get("nullable", False),
                "default": spec.get("default"),
            }
            if spec.get("schema"):
                params["schema"] = spec.get("schema")
            else:
                params["schema"] = {"type": spec["type"]}
            if spec.get("items"):
                params["schema"]["items"] = spec.get("items")
            if "enum" in spec:
                params["schema"]["enum"] = spec["enum"]

            parameters.append(params)

            if spec["type"] == "array":
                # To correctly handle array into the url query string,
                # the name must ends with []
                params["name"] = params["name"] + "[]"

        return parameters

    def to_openapi_requestbody(self, service):
        json_schema = self.to_json_schema(service, "input")
        return {"content": {"application/json": {"schema": json_schema}}}

    def to_openapi_responses(self, service):
        json_schema = self.to_json_schema(service, "output")
        return {"200": {"content": {"application/json": {"schema": json_schema}}}}

    def get_cerberus_validator(self, service, direction):
        assert direction in ("input", "output")
        schema = self._schema
        if isinstance(self._schema, str):
            validator_component = service.component(usage="cerberus.validator")
            schema = validator_component.get_validator_handler(
                service, self._schema, direction
            )()
        if isinstance(schema, Validator):
            return schema
        if isinstance(schema, dict):
            return Validator(schema, purge_unknown=True)
        raise Exception(_("Unable to get cerberus schema from %s") % self._schema)

    def to_json_schema(self, service, direction):
        schema = self.get_cerberus_validator(service, direction).schema
        return cerberus_to_json(schema)


class CerberusListValidator(CerberusValidator):
    def __init__(self, schema, min_items=None, max_items=None, unique_items=None):
        """
        :param schema: Cerberus list item schema
                       can be dict as cerberus schema, an instance of
                       cerberus.Validator or a sting with the method name to
                       call on the service to get the schema or the validator
        :param min_items: A list instance is valid against "min_items" if its
                          size is greater than, or equal to, min_items.
                          The value MUST be a non-negative integer.
        :param max_items: A list instance is valid against "max_items" if its
                          size is less than, or equal to, max_items.
                          The value MUST be a non-negative integer.
        :param unique_items: Used to document that the list should only
                             contain unique items.
                             (Not enforced at validation time)
        """
        super(CerberusListValidator, self).__init__(schema=schema)
        self._min_items = min_items
        self._max_items = max_items
        self._unique_items = unique_items

    def from_params(self, service, params):
        return self._do_validate(service, data=params, direction="input")

    def to_response(self, service, result):
        return self._do_validate(service, data=result, direction="output")

    def to_openapi_query_parameters(self, service):
        raise NotImplementedError("List are not (?yet?) supported as query paramters")

    def _do_validate(self, service, data, direction):
        validator = self.get_cerberus_validator(service, direction)
        values = []
        ExceptionClass = UserError if direction == "input" else SystemError
        for idx, p in enumerate(data):
            if not validator.validate(p):
                raise ExceptionClass(
                    _("BadRequest item %s :%s") % (idx, validator.errors)
                )
            values.append(validator.document)
        if self._min_items is not None and len(values) < self._min_items:
            raise ExceptionClass(
                _(
                    "BadRequest: Not enough items in the list (%s < %s)"
                    % (len(values), self._min_items)
                )
            )
        if self._max_items is not None and len(values) > self._max_items:
            raise ExceptionClass(
                _(
                    "BadRequest: Too many items in the list (%s > %s)"
                    % (len(values), self._max_items)
                )
            )
        return values

    def to_json_schema(self, service, direction):
        cerberus_schema = self.get_cerberus_validator(service, direction).schema
        json_schema = cerberus_to_json(cerberus_schema)
        json_schema = {"type": "array", "items": json_schema}
        if self._min_items is not None:
            json_schema["minItems"] = self._min_items
        if self._max_items is not None:
            json_schema["maxItems"] = self._max_items
        if self._unique_items is not None:
            json_schema["uniqueItems"] = self._unique_items
        return json_schema