""" This module defines an AbstractAPI, which defines a standardized interface for a Connexion API. """ import abc import logging import pathlib import sys import typing as t import warnings from enum import Enum from ..decorators.produces import NoContent from ..exceptions import ResolverError from ..http_facts import METHODS from ..jsonifier import Jsonifier from ..lifecycle import ConnexionResponse from ..operations import make_operation from ..options import ConnexionOptions from ..resolver import Resolver from ..spec import Specification from ..utils import is_json_mimetype MODULE_PATH = pathlib.Path(__file__).absolute().parent.parent SWAGGER_UI_URL = 'ui' logger = logging.getLogger('connexion.apis.abstract') class AbstractAPIMeta(abc.ABCMeta): def __init__(cls, name, bases, attrs): abc.ABCMeta.__init__(cls, name, bases, attrs) cls._set_jsonifier() class AbstractAPI(metaclass=AbstractAPIMeta): """ Defines an abstract interface for a Swagger API """ def __init__(self, specification, base_path=None, arguments=None, validate_responses=False, strict_validation=False, resolver=None, auth_all_paths=False, debug=False, resolver_error_handler=None, validator_map=None, pythonic_params=False, pass_context_arg_name=None, options=None, ): """ :type specification: pathlib.Path | dict :type base_path: str | None :type arguments: dict | None :type validate_responses: bool :type strict_validation: bool :type auth_all_paths: bool :type debug: bool :param validator_map: Custom validators for the types "parameter", "body" and "response". :type validator_map: dict :param resolver: Callable that maps operationID to a function :param resolver_error_handler: If given, a callable that generates an Operation used for handling ResolveErrors :type resolver_error_handler: callable | None :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 options: New style options dictionary. :type options: dict | None :param pass_context_arg_name: If not None URL request handling functions with an argument matching this name will be passed the framework's request context. :type pass_context_arg_name: str | None """ self.debug = debug self.validator_map = validator_map self.resolver_error_handler = resolver_error_handler logger.debug('Loading specification: %s', specification, extra={'swagger_yaml': specification, 'base_path': base_path, 'arguments': arguments, 'auth_all_paths': auth_all_paths}) # Avoid validator having ability to modify specification self.specification = Specification.load(specification, arguments=arguments) logger.debug('Read specification', extra={'spec': self.specification}) self.options = ConnexionOptions(options, oas_version=self.specification.version) logger.debug('Options Loaded', extra={'swagger_ui': self.options.openapi_console_ui_available, 'swagger_path': self.options.openapi_console_ui_from_dir, 'swagger_url': self.options.openapi_console_ui_path}) self._set_base_path(base_path) logger.debug('Security Definitions: %s', self.specification.security_definitions) self.resolver = resolver or Resolver() logger.debug('Validate Responses: %s', str(validate_responses)) self.validate_responses = validate_responses logger.debug('Strict Request Validation: %s', str(strict_validation)) self.strict_validation = strict_validation logger.debug('Pythonic params: %s', str(pythonic_params)) self.pythonic_params = pythonic_params logger.debug('pass_context_arg_name: %s', pass_context_arg_name) self.pass_context_arg_name = pass_context_arg_name self.security_handler_factory = self.make_security_handler_factory(pass_context_arg_name) if self.options.openapi_spec_available: self.add_openapi_json() self.add_openapi_yaml() if self.options.openapi_console_ui_available: self.add_swagger_ui() self.add_paths() if auth_all_paths: self.add_auth_on_not_found( self.specification.security, self.specification.security_definitions ) def _set_base_path(self, base_path=None): if base_path is not None: # update spec to include user-provided base_path self.specification.base_path = base_path self.base_path = base_path else: self.base_path = self.specification.base_path @abc.abstractmethod def add_openapi_json(self): """ Adds openapi spec to {base_path}/openapi.json (or {base_path}/swagger.json for swagger2) """ @abc.abstractmethod def add_swagger_ui(self): """ Adds swagger ui to {base_path}/ui/ """ @abc.abstractmethod def add_auth_on_not_found(self, security, security_definitions): """ Adds a 404 error handler to authenticate and only expose the 404 status if the security validation pass. """ @staticmethod @abc.abstractmethod def make_security_handler_factory(pass_context_arg_name): """ Create SecurityHandlerFactory to create all security check handlers """ def add_operation(self, path, method): """ Adds one operation to the api. This method uses the OperationID identify the module and function that will handle the operation From Swagger Specification: **OperationID** A friendly name for the operation. The id MUST be unique among all operations described in the API. Tools and libraries MAY use the operation id to uniquely identify an operation. :type method: str :type path: str """ operation = make_operation( self.specification, self, path, method, self.resolver, validate_responses=self.validate_responses, validator_map=self.validator_map, strict_validation=self.strict_validation, pythonic_params=self.pythonic_params, uri_parser_class=self.options.uri_parser_class, pass_context_arg_name=self.pass_context_arg_name ) self._add_operation_internal(method, path, operation) @abc.abstractmethod def _add_operation_internal(self, method, path, operation): """ Adds the operation according to the user framework in use. It will be used to register the operation on the user framework router. """ def _add_resolver_error_handler(self, method, path, err): """ Adds a handler for ResolverError for the given method and path. """ operation = self.resolver_error_handler( err, security=self.specification.security, security_definitions=self.specification.security_definitions ) self._add_operation_internal(method, path, operation) def add_paths(self, paths=None): """ Adds the paths defined in the specification as endpoints :type paths: list """ paths = paths or self.specification.get('paths', dict()) for path, methods in paths.items(): logger.debug('Adding %s%s...', self.base_path, path) for method in methods: if method not in METHODS: continue try: self.add_operation(path, method) except ResolverError as err: # If we have an error handler for resolver errors, add it as an operation. # Otherwise treat it as any other error. if self.resolver_error_handler is not None: self._add_resolver_error_handler(method, path, err) else: self._handle_add_operation_error(path, method, err.exc_info) except Exception: # All other relevant exceptions should be handled as well. self._handle_add_operation_error(path, method, sys.exc_info()) def _handle_add_operation_error(self, path, method, exc_info): url = f'{self.base_path}{path}' error_msg = 'Failed to add operation for {method} {url}'.format( method=method.upper(), url=url) if self.debug: logger.exception(error_msg) else: logger.error(error_msg) _type, value, traceback = exc_info raise value.with_traceback(traceback) @classmethod @abc.abstractmethod def get_request(self, *args, **kwargs): """ This method converts the user framework request to a ConnexionRequest. """ @classmethod @abc.abstractmethod def get_response(self, response, mimetype=None, request=None): """ This method converts a handler response to a framework response. This method should just retrieve response from handler then call `cls._get_response`. It is mainly here to handle AioHttp async handler. :param response: A response to cast (tuple, framework response, etc). :param mimetype: The response mimetype. :type mimetype: Union[None, str] :param request: The request associated with this response (the user framework request). """ @classmethod def _get_response(cls, response, mimetype=None, extra_context=None): """ This method converts a handler response to a framework response. The response can be a ConnexionResponse, an operation handler, a framework response or a tuple. Other type than ConnexionResponse are handled by `cls._response_from_handler` :param response: A response to cast (tuple, framework response, etc). :param mimetype: The response mimetype. :type mimetype: Union[None, str] :param extra_context: dict of extra details, like url, to include in logs :type extra_context: Union[None, dict] """ if extra_context is None: extra_context = {} logger.debug('Getting data and status code', extra={ 'data': response, 'data_type': type(response), **extra_context }) if isinstance(response, ConnexionResponse): framework_response = cls._connexion_to_framework_response(response, mimetype, extra_context) else: framework_response = cls._response_from_handler(response, mimetype, extra_context) logger.debug('Got framework response', extra={ 'response': framework_response, 'response_type': type(framework_response), **extra_context }) return framework_response @classmethod def _response_from_handler( cls, response: t.Union[t.Any, str, t.Tuple[str], t.Tuple[str, int], t.Tuple[str, int, dict]], mimetype: str, extra_context: t.Optional[dict] = None ) -> t.Any: """ Create a framework response from the operation handler data. An operation handler can return: - a framework response - a body (str / binary / dict / list), a response will be created with a status code 200 by default and empty headers. - a tuple of (body: str, status_code: int) - a tuple of (body: str, status_code: int, headers: dict) :param response: A response from an operation handler. :param mimetype: The response mimetype. :param extra_context: dict of extra details, like url, to include in logs """ if cls._is_framework_response(response): return response if isinstance(response, tuple): len_response = len(response) if len_response == 1: data, = response return cls._build_response(mimetype=mimetype, data=data, extra_context=extra_context) if len_response == 2: if isinstance(response[1], (int, Enum)): data, status_code = response return cls._build_response(mimetype=mimetype, data=data, status_code=status_code, extra_context=extra_context) else: data, headers = response return cls._build_response(mimetype=mimetype, data=data, headers=headers, extra_context=extra_context) elif len_response == 3: data, status_code, headers = response return cls._build_response(mimetype=mimetype, data=data, status_code=status_code, headers=headers, extra_context=extra_context) else: raise TypeError( 'The view function did not return a valid response tuple.' ' The tuple must have the form (body), (body, status, headers),' ' (body, status), or (body, headers).' ) else: return cls._build_response(mimetype=mimetype, data=response, extra_context=extra_context) @classmethod def get_connexion_response(cls, response, mimetype=None): """ Cast framework dependent response to ConnexionResponse used for schema validation """ if isinstance(response, ConnexionResponse): # If body in ConnexionResponse is not byte, it may not pass schema validation. # In this case, rebuild response with aiohttp to have consistency if response.body is None or isinstance(response.body, bytes): return response else: response = cls._build_response( data=response.body, mimetype=mimetype, content_type=response.content_type, headers=response.headers, status_code=response.status_code ) if not cls._is_framework_response(response): response = cls._response_from_handler(response, mimetype) return cls._framework_to_connexion_response(response=response, mimetype=mimetype) @classmethod @abc.abstractmethod def _is_framework_response(cls, response): """ Return True if `response` is a framework response class """ @classmethod @abc.abstractmethod def _framework_to_connexion_response(cls, response, mimetype): """ Cast framework response class to ConnexionResponse used for schema validation """ @classmethod @abc.abstractmethod def _connexion_to_framework_response(cls, response, mimetype, extra_context=None): """ Cast ConnexionResponse to framework response class """ @classmethod @abc.abstractmethod def _build_response(cls, data, mimetype, content_type=None, status_code=None, headers=None, extra_context=None): """ Create a framework response from the provided arguments. :param data: Body data. :param content_type: The response mimetype. :type content_type: str :param content_type: The response status code. :type status_code: int :param headers: The response status code. :type headers: Union[Iterable[Tuple[str, str]], Dict[str, str]] :param extra_context: dict of extra details, like url, to include in logs :type extra_context: Union[None, dict] :return A framework response. :rtype Response """ @classmethod def _prepare_body_and_status_code(cls, data, mimetype, status_code=None, extra_context=None): if data is NoContent: data = None if status_code is None: if data is None: status_code = 204 mimetype = None else: status_code = 200 elif hasattr(status_code, "value"): # If we got an enum instead of an int, extract the value. status_code = status_code.value if data is not None: body, mimetype = cls._serialize_data(data, mimetype) else: body = data if extra_context is None: extra_context = {} logger.debug('Prepared body and status code (%d)', status_code, extra={ 'body': body, **extra_context }) return body, status_code, mimetype @classmethod def _serialize_data(cls, data, mimetype): # TODO: Harmonize with flask_api. Currently this is the backwards compatible with aiohttp_api._cast_body. if not isinstance(data, bytes): if isinstance(mimetype, str) and is_json_mimetype(mimetype): body = cls.jsonifier.dumps(data) elif isinstance(data, str): body = data else: warnings.warn( "Implicit (aiohttp) serialization with str() will change in the next major version. " "This is triggered because a non-JSON response body is being stringified. " "This will be replaced by something that is mimetype-specific and may " "serialize some things as JSON or throw an error instead of silently " "stringifying unknown response bodies. " "Please make sure to specify media/mime types in your specs.", FutureWarning # a Deprecation targeted at application users. ) body = str(data) else: body = data return body, mimetype def json_loads(self, data): return self.jsonifier.loads(data) @classmethod def _set_jsonifier(cls): cls.jsonifier = Jsonifier()