airflow_scheduler/.venv/Lib/site-packages/connexion/operations/abstract.py

"""
This module defines an AbstractOperation class which implements an abstract Operation interface
and functionality shared between Swagger 2 and OpenAPI 3 specifications.
"""

import abc
import logging

from connexion.operations.secure import SecureOperation

from ..decorators.metrics import UWSGIMetricsCollector
from ..decorators.parameter import parameter_to_arg
from ..decorators.produces import BaseSerializer, Produces
from ..decorators.response import ResponseValidator
from ..decorators.validation import ParameterValidator, RequestBodyValidator
from ..utils import all_json, is_nullable

logger = logging.getLogger('connexion.operations.abstract')

DEFAULT_MIMETYPE = 'application/json'

VALIDATOR_MAP = {
    'parameter': ParameterValidator,
    'body': RequestBodyValidator,
    'response': ResponseValidator,
}


class AbstractOperation(SecureOperation, metaclass=abc.ABCMeta):

    """
    An API routes requests to an Operation by a (path, method) pair.
    The operation uses a resolver to resolve its handler function.
    We use the provided spec to do a bunch of heavy lifting before
    (and after) we call security_schemes handler.
    The registered handler function ends up looking something like::

        @secure_endpoint
        @validate_inputs
        @deserialize_function_inputs
        @serialize_function_outputs
        @validate_outputs
        def user_provided_handler_function(important, stuff):
            if important:
                serious_business(stuff)
    """
    def __init__(self, api, method, path, operation, resolver,
                 app_security=None, security_schemes=None,
                 validate_responses=False, strict_validation=False,
                 randomize_endpoint=None, validator_map=None,
                 pythonic_params=False, uri_parser_class=None,
                 pass_context_arg_name=None):
        """
        :param api: api that this operation is attached to
        :type api: apis.AbstractAPI
        :param method: HTTP method
        :type method: str
        :param path:
        :type path: str
        :param operation: swagger operation object
        :type operation: dict
        :param resolver: Callable that maps operationID to a function
        :param app_produces: list of content types the application can return by default
        :param app_security: list of security rules the application uses by default
        :type app_security: list
        :param security_schemes: `Security Definitions Object
            <https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#security-definitions-object>`_
        :type security_schemes: dict
        :param validate_responses: True enables validation. Validation errors generate HTTP 500 responses.
        :type validate_responses: bool
        :param strict_validation: True enables validation on invalid request parameters
        :type strict_validation: bool
        :param randomize_endpoint: number of random characters to append to operation name
        :type randomize_endpoint: integer
        :param validator_map: Custom validators for the types "parameter", "body" and "response".
        :type validator_map: dict
        :param pythonic_params: When True CamelCase parameters are converted to snake_case and an underscore is appended
            to any shadowed built-ins
        :type pythonic_params: bool
        :param uri_parser_class: class to use for uri parsing
        :type uri_parser_class: AbstractURIParser
        :param pass_context_arg_name: If not None will try to inject the request context to the function using this
            name.
        :type pass_context_arg_name: str|None
        """
        self._api = api
        self._method = method
        self._path = path
        self._operation = operation
        self._resolver = resolver
        self._security = app_security
        self._security_schemes = security_schemes
        self._validate_responses = validate_responses
        self._strict_validation = strict_validation
        self._pythonic_params = pythonic_params
        self._uri_parser_class = uri_parser_class
        self._pass_context_arg_name = pass_context_arg_name
        self._randomize_endpoint = randomize_endpoint

        self._operation_id = self._operation.get("operationId")
        self._resolution = resolver.resolve(self)
        self._operation_id = self._resolution.operation_id

        self._responses = self._operation.get("responses", {})

        self._validator_map = dict(VALIDATOR_MAP)
        self._validator_map.update(validator_map or {})

    @property
    def method(self):
        """
        The HTTP method for this operation (ex. GET, POST)
        """
        return self._method

    @property
    def path(self):
        """
        The path of the operation, relative to the API base path
        """
        return self._path

    @property
    def responses(self):
        """
        Returns the responses for this operation
        """
        return self._responses

    @property
    def validator_map(self):
        """
        Validators to use for parameter, body, and response validation
        """
        return self._validator_map

    @property
    def operation_id(self):
        """
        The operation id used to identify the operation internally to the app
        """
        return self._operation_id

    @property
    def randomize_endpoint(self):
        """
        number of random digits to generate and append to the operation_id.
        """
        return self._randomize_endpoint

    @property
    def router_controller(self):
        """
        The router controller to use (python module where handler functions live)
        """
        return self._router_controller

    @property
    def strict_validation(self):
        """
        If True, validate all requests against the spec
        """
        return self._strict_validation

    @property
    def pythonic_params(self):
        """
        If True, convert CamelCase into pythonic_variable_names
        """
        return self._pythonic_params

    @property
    def validate_responses(self):
        """
        If True, check the response against the response schema, and return an
        error if the response does not validate.
        """
        return self._validate_responses

    @staticmethod
    def _get_file_arguments(files, arguments, has_kwargs=False):
        return {k: v for k, v in files.items() if k in arguments or has_kwargs}

    @abc.abstractmethod
    def _get_val_from_param(self, value, query_defn):
        """
        Convert input parameters into the correct type
        """

    def _query_args_helper(self, query_defns, query_arguments,
                           function_arguments, has_kwargs, sanitize):
        res = {}
        for key, value in query_arguments.items():
            sanitized_key = sanitize(key)
            if not has_kwargs and sanitized_key not in function_arguments:
                logger.debug("Query Parameter '%s' (sanitized: '%s') not in function arguments",
                             key, sanitized_key)
            else:
                logger.debug("Query Parameter '%s' (sanitized: '%s') in function arguments",
                             key, sanitized_key)
                try:
                    query_defn = query_defns[key]
                except KeyError:  # pragma: no cover
                    logger.error("Function argument '%s' (non-sanitized: %s) not defined in specification",
                                 sanitized_key, key)
                else:
                    logger.debug('%s is a %s', key, query_defn)
                    res.update({sanitized_key: self._get_val_from_param(value, query_defn)})
        return res

    @abc.abstractmethod
    def _get_query_arguments(self, query, arguments, has_kwargs, sanitize):
        """
        extract handler function arguments from the query parameters
        """

    @abc.abstractmethod
    def _get_body_argument(self, body, arguments, has_kwargs, sanitize):
        """
        extract handler function arguments from the request body
        """

    def _get_path_arguments(self, path_params, sanitize):
        """
        extract handler function arguments from path parameters
        """
        kwargs = {}
        path_defns = {p["name"]: p for p in self.parameters if p["in"] == "path"}
        for key, value in path_params.items():
            sanitized_key = sanitize(key)
            if key in path_defns:
                kwargs[sanitized_key] = self._get_val_from_param(value, path_defns[key])
            else:  # Assume path params mechanism used for injection
                kwargs[sanitized_key] = value
        return kwargs

    @property
    @abc.abstractmethod
    def parameters(self):
        """
        Returns the parameters for this operation
        """

    @property
    @abc.abstractmethod
    def produces(self):
        """
        Content-Types that the operation produces
        """

    @property
    @abc.abstractmethod
    def consumes(self):
        """
        Content-Types that the operation consumes
        """

    @property
    @abc.abstractmethod
    def body_schema(self):
        """
        The body schema definition for this operation.
        """

    @property
    @abc.abstractmethod
    def body_definition(self):
        """
        The body definition for this operation.
        :rtype: dict
        """

    def get_arguments(self, path_params, query_params, body, files, arguments,
                      has_kwargs, sanitize):
        """
        get arguments for handler function
        """
        ret = {}
        ret.update(self._get_path_arguments(path_params, sanitize))
        ret.update(self._get_query_arguments(query_params, arguments,
                                             has_kwargs, sanitize))

        if self.method.upper() in ["PATCH", "POST", "PUT"]:
            ret.update(self._get_body_argument(body, arguments,
                                               has_kwargs, sanitize))
            ret.update(self._get_file_arguments(files, arguments, has_kwargs))
        return ret

    def response_definition(self, status_code=None,
                            content_type=None):
        """
        response definition for this endpoint
        """
        content_type = content_type or self.get_mimetype()
        response_definition = self.responses.get(
            str(status_code),
            self.responses.get("default", {})
        )
        return response_definition

    @abc.abstractmethod
    def response_schema(self, status_code=None, content_type=None):
        """
        response schema for this endpoint
        """

    @abc.abstractmethod
    def example_response(self, status_code=None, content_type=None):
        """
        Returns an example from the spec
        """

    @abc.abstractmethod
    def get_path_parameter_types(self):
        """
        Returns the types for parameters in the path
        """

    @abc.abstractmethod
    def with_definitions(self, schema):
        """
        Returns the given schema, but with the definitions from the spec
        attached. This allows any remaining references to be resolved by a
        validator (for example).
        """

    def get_mimetype(self):
        """
        If the endpoint has no 'produces' then the default is
        'application/json'.

        :rtype str
        """
        if all_json(self.produces):
            try:
                return self.produces[0]
            except IndexError:
                return DEFAULT_MIMETYPE
        elif len(self.produces) == 1:
            return self.produces[0]
        else:
            return DEFAULT_MIMETYPE

    @property
    def _uri_parsing_decorator(self):
        """
        Returns a decorator that parses request data and handles things like
        array types, and duplicate parameter definitions.
        """
        return self._uri_parser_class(self.parameters, self.body_definition)

    @property
    def function(self):
        """
        Operation function with decorators

        :rtype: types.FunctionType
        """
        function = parameter_to_arg(
            self, self._resolution.function, self.pythonic_params,
            self._pass_context_arg_name
        )

        if self.validate_responses:
            logger.debug('... Response validation enabled.')
            response_decorator = self.__response_validation_decorator
            logger.debug('... Adding response decorator (%r)', response_decorator)
            function = response_decorator(function)

        produces_decorator = self.__content_type_decorator
        logger.debug('... Adding produces decorator (%r)', produces_decorator)
        function = produces_decorator(function)

        for validation_decorator in self.__validation_decorators:
            function = validation_decorator(function)

        uri_parsing_decorator = self._uri_parsing_decorator
        function = uri_parsing_decorator(function)

        # NOTE: the security decorator should be applied last to check auth before anything else :-)
        security_decorator = self.security_decorator
        logger.debug('... Adding security decorator (%r)', security_decorator)
        function = security_decorator(function)

        function = self._request_response_decorator(function)

        if UWSGIMetricsCollector.is_available():  # pragma: no cover
            decorator = UWSGIMetricsCollector(self.path, self.method)
            function = decorator(function)

        return function

    @property
    def __content_type_decorator(self):
        """
        Get produces decorator.

        If the operation mimetype format is json then the function return value is jsonified

        From Swagger Specification:

        **Produces**

        A list of MIME types the operation can produce. This overrides the produces definition at the Swagger Object.
        An empty value MAY be used to clear the global definition.

        :rtype: types.FunctionType
        """

        logger.debug('... Produces: %s', self.produces, extra=vars(self))

        mimetype = self.get_mimetype()
        if all_json(self.produces):  # endpoint will return json
            logger.debug('... Produces json', extra=vars(self))
            # TODO: Refactor this.
            return lambda f: f

        elif len(self.produces) == 1:
            logger.debug('... Produces %s', mimetype, extra=vars(self))
            decorator = Produces(mimetype)
            return decorator

        else:
            return BaseSerializer()

    @property
    def __validation_decorators(self):
        """
        :rtype: types.FunctionType
        """
        ParameterValidator = self.validator_map['parameter']
        RequestBodyValidator = self.validator_map['body']
        if self.parameters:
            yield ParameterValidator(self.parameters,
                                     self.api,
                                     strict_validation=self.strict_validation)
        if self.body_schema:
            yield RequestBodyValidator(self.body_schema, self.consumes, self.api,
                                       is_nullable(self.body_definition),
                                       strict_validation=self.strict_validation)

    @property
    def __response_validation_decorator(self):
        """
        Get a decorator for validating the generated Response.
        :rtype: types.FunctionType
        """
        ResponseValidator = self.validator_map['response']
        return ResponseValidator(self, self.get_mimetype())

    def json_loads(self, data):
        """
        A wrapper for calling the API specific JSON loader.

        :param data: The JSON data in textual form.
        :type data: bytes
        """
        return self.api.json_loads(data)