Skip to content

Commit

Permalink
Refactor decorators (#1618)
Browse files Browse the repository at this point in the history 
This PR refactors the decorators so they can be used independently of an
API, which will allow them to be used by other wrapped frameworks as
well.
  • Loading branch information
@RobbeSneyders
RobbeSneyders committed Jan 7, 2023
2 parents 073f0d4 + de3b6f6 commit 9067287
Show file tree
Hide file tree
Showing 27 changed files with 906 additions and 762 deletions.
218 changes: 3 additions & 215 deletions connexion/apis/abstract.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,35 +6,25 @@
import pathlib
import sys
import typing as t
from enum import Enum

from starlette.requests import Request

from connexion.datastructures import NoContent
from connexion.decorators.parameter import parameter_to_arg
from connexion.exceptions import ResolverError
from connexion.http_facts import METHODS
from connexion.jsonifier import Jsonifier
from connexion.lifecycle import ConnexionResponse
from connexion.operations import make_operation
from connexion.options import ConnexionOptions
from connexion.resolver import Resolver
from connexion.spec import Specification
from connexion.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 AbstractSpecAPI:

jsonifier = Jsonifier()

class AbstractSpecAPI(metaclass=AbstractAPIMeta):
def __init__(
self,
specification: t.Union[pathlib.Path, str, dict],
Expand Down Expand Up @@ -94,10 +84,6 @@ def _set_base_path(self, base_path: t.Optional[str] = None) -> None:
else:
self.base_path = self.specification.base_path

@classmethod
def _set_jsonifier(cls):
cls.jsonifier = Jsonifier()


class AbstractRoutingAPI(AbstractSpecAPI):
def __init__(
Expand Down Expand Up @@ -173,7 +159,7 @@ def _handle_add_operation_error(self, path: str, method: str, exc_info: tuple):
raise value.with_traceback(traceback)


class AbstractAPI(AbstractRoutingAPI, metaclass=AbstractAPIMeta):
class AbstractAPI(AbstractRoutingAPI):
"""
Defines an abstract interface for a Swagger API
"""
Expand Down Expand Up @@ -225,207 +211,9 @@ def add_operation(self, path, method):
path,
method,
self.resolver,
pythonic_params=self.pythonic_params,
uri_parser_class=self.options.uri_parser_class,
parameter_to_arg=parameter_to_arg,
)
self._add_operation_internal(method, path, operation)

@classmethod
@abc.abstractmethod
def get_request(cls, **kwargs) -> Request:
"""
This method converts the user framework request to a ConnexionRequest.
"""

@classmethod
@abc.abstractmethod
def get_response(cls, response, mimetype=None):
"""
This method converts a handler response to a framework response.
This method should just retrieve response from handler then call `cls._get_response`.
:param response: A response to cast (tuple, framework response, etc).
:param mimetype: The response mimetype.
:type mimetype: Union[None, str]
"""

@classmethod
def _get_response(cls, response, mimetype=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]
"""
logger.debug(
"Getting data and status code",
extra={"data": response, "data_type": type(response)},
)

if isinstance(response, ConnexionResponse):
framework_response = cls._connexion_to_framework_response(
response, mimetype
)
else:
framework_response = cls._response_from_handler(response, mimetype)

logger.debug(
"Got framework response",
extra={
"response": framework_response,
"response_type": type(framework_response),
},
)
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,
) -> 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.
"""
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)
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,
)
else:
data, headers = response
return cls._build_response(
mimetype=mimetype,
data=data,
headers=headers,
)
elif len_response == 3:
data, status_code, headers = response
return cls._build_response(
mimetype=mimetype,
data=data,
status_code=status_code,
headers=headers,
)
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)

@classmethod
def get_connexion_response(cls, response, mimetype=None):
"""Cast framework dependent response to ConnexionResponse used for schema validation"""
if isinstance(response, ConnexionResponse):
return response

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):
"""Cast ConnexionResponse to framework response class"""

@classmethod
@abc.abstractmethod
def _build_response(
cls,
data,
mimetype,
content_type=None,
status_code=None,
headers=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]]
:return A framework response.
:rtype Response
"""

@classmethod
def _prepare_body_and_status_code(cls, data, mimetype, status_code=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

logger.debug(
"Prepared body and status code (%d)",
status_code,
extra={"body": body},
)

return body, status_code, mimetype

@classmethod
def _serialize_data(cls, data, mimetype):
if isinstance(mimetype, str) and is_json_mimetype(mimetype):
body = cls.jsonifier.dumps(data)
else:
body = data

return body, mimetype

def json_loads(self, data):
return self.jsonifier.loads(data)

0 comments on commit 9067287

Please sign in to comment.