Source code for magpie.api.exception

import json
import platform
import re
from sys import exc_info
from typing import TYPE_CHECKING

import colander
import six
from dicttoxml import dicttoxml
from pyramid.httpexceptions import (
    HTTPBadRequest,
    HTTPError,
    HTTPException,
    HTTPInternalServerError,
    HTTPOk,
    HTTPRedirection,
    HTTPSuccessful
)

from magpie.utils import (
    CONTENT_TYPE_ANY,
    CONTENT_TYPE_APP_XML,
    CONTENT_TYPE_HTML,
    CONTENT_TYPE_JSON,
    CONTENT_TYPE_PLAIN,
    CONTENT_TYPE_TXT_XML,
    SUPPORTED_ACCEPT_TYPES,
    get_header,
    get_logger,
    isclass,
    islambda
)

if TYPE_CHECKING:
    # pylint: disable=W0611,unused-import
    from typing import Any, Callable, Iterable, List, NoReturn, Optional, Tuple, Type, Union

    from magpie.typedefs import JSON, ParamsType, Str

[docs] LOGGER = get_logger(__name__)
# control variables to avoid infinite recursion in case of # major programming error to avoid application hanging
[docs] RAISE_RECURSIVE_SAFEGUARD_MAX = 5
[docs] RAISE_RECURSIVE_SAFEGUARD_COUNT = 0
# utility parameter validation regexes for 'matches' argument
[docs] PARAM_REGEX = re.compile(r"^[A-Za-z0-9]+(?:[\s_\-\.][A-Za-z0-9]+)*$") # request parameters
[docs] SCOPE_REGEX = re.compile(r"^[A-Za-z0-9]+(?:[\:\s_\-\.][A-Za-z0-9]+)*$") # allow scoped names (e.g.: 'namespace:value')
[docs] EMAIL_REGEX = re.compile(colander.EMAIL_RE)
[docs] UUID_REGEX = re.compile(colander.UUID_REGEX)
[docs] URL_REGEX = re.compile(colander.URL_REGEX, re.I | re.X)
[docs] INDEX_REGEX = re.compile(r"^[0-9]+$")
if platform.python_version() >= "3.7":
[docs] Pattern = re.Pattern
else: Pattern = type(re.compile(""))
[docs] def verify_param( # noqa: E126 # pylint: disable=R0913,too-many-arguments # --- verification values --- # noqa: E126 param, # type: Any param_compare=None, # type: Optional[Union[Any, List[Any]]] # --- output options on failure --- param_name=None, # type: Optional[Str] param_content=None, # type: Optional[JSON] with_param=True, # type: bool http_error=HTTPBadRequest, # type: Type[HTTPError] http_kwargs=None, # type: Optional[ParamsType] msg_on_fail="", # type: Str content=None, # type: Optional[JSON] content_type=CONTENT_TYPE_JSON, # type: Str metadata=None, # type: Optional[JSON] # --- verification flags (method) --- not_none=False, # type: bool not_empty=False, # type: bool not_in=False, # type: bool not_equal=False, # type: bool is_true=False, # type: bool is_false=False, # type: bool is_none=False, # type: bool is_empty=False, # type: bool is_in=False, # type: bool is_equal=False, # type: bool is_type=False, # type: bool matches=False, # type: bool ): # type: (...) -> None # noqa: E123,E126 # pylint: disable=R0912,R0914 """ Evaluate various parameter combinations given the requested verification flags. Given a failing verification, directly raises the specified :paramref:`http_error`. Invalid usage exceptions generated by this verification process are treated as :class:`HTTPInternalServerError`. Exceptions are generated using the standard output method. :param param: parameter value to evaluate :param param_compare: Other value(s) to test :paramref:`param` against. Can be an iterable (single value resolved as iterable unless ``None``). To test for ``None`` type, use :paramref:`is_none`/:paramref:`not_none` flags instead. :param param_name: name of the tested parameter returned in response if specified for debugging purposes :param param_content: Additional JSON content to apply to generated error content on raise when :paramref:`with_param` is ``True``. Must be JSON serializable. Provided content can override generated error parameter if matching fields. :param with_param: On raise, adds values of :paramref:`param`, :paramref:`param_name` and :paramref:`param_compare`, as well as additional failing conditions metadata to the JSON response body for each of the corresponding value. :param http_error: derived exception to raise on test failure (default: :class:`HTTPBadRequest`) :param http_kwargs: additional keyword arguments to pass to :paramref:`http_error` called in case of HTTP exception :param msg_on_fail: message details to return in HTTP exception if flag condition failed :param content: json formatted additional content to provide in case of exception :param content_type: format in which to return the exception (one of :py:data:`magpie.common.SUPPORTED_ACCEPT_TYPES`) :param metadata: request metadata to add to the response body. (see: :func:`magpie.api.requests.get_request_info`) :param not_none: test that :paramref:`param` is not ``None`` type :param not_empty: test that :paramref:`param` is not an empty iterable (string, list, set, etc.) :param not_in: test that :paramref:`param` does not exist in :paramref:`param_compare` values :param not_equal: test that :paramref:`param` is not equal to :paramref:`param_compare` value :param is_true: test that :paramref:`param` is ``True`` :param is_false: test that :paramref:`param` is ``False`` :param is_none: test that :paramref:`param` is ``None`` type :param is_empty: test `param` for an empty iterable (string, list, set, etc.) :param is_in: test that :paramref:`param` exists in :paramref:`param_compare` values :param is_equal: test that :paramref:`param` equals :paramref:`param_compare` value :param is_type: test that :paramref:`param` is of same type as specified by :paramref:`param_compare` type :param matches: test that :paramref:`param` matches the regex specified by :paramref:`param_compare` value :raises HTTPError: if tests fail, specified exception is raised (default: :class:`HTTPBadRequest`) :raises HTTPInternalServerError: for evaluation error :return: nothing if all tests passed """ content = {} if content is None else content needs_compare = is_type or is_in or not_in or is_equal or not_equal or matches needs_iterable = is_in or not_in # precondition evaluation of input parameters try: # following TypeError/ValueError are used instead of HTTPError as they would be incorrect setup by the developer # after validation of their conditions, we do actual validation of the parameters according to conditions if not isinstance(not_none, bool): raise TypeError("'not_none' is not a 'bool'") if not isinstance(not_empty, bool): raise TypeError("'not_empty' is not a 'bool'") if not isinstance(not_in, bool): raise TypeError("'not_in' is not a 'bool'") if not isinstance(not_equal, bool): raise TypeError("'not_equal' is not a 'bool'") if not isinstance(is_true, bool): raise TypeError("'is_true' is not a 'bool'") if not isinstance(is_false, bool): raise TypeError("'is_false' is not a 'bool'") if not isinstance(is_none, bool): raise TypeError("'is_none' is not a 'bool'") if not isinstance(is_empty, bool): raise TypeError("'is_empty' is not a 'bool'") if not isinstance(is_in, bool): raise TypeError("'is_in' is not a 'bool'") if not isinstance(is_equal, bool): raise TypeError("'is_equal' is not a 'bool'") if not isinstance(is_type, bool): raise TypeError("'is_type' is not a 'bool'") if not isinstance(matches, bool): raise TypeError("'matches' is not a 'bool'") # error if none of the flags specified if not any([not_none, not_empty, not_in, not_equal, is_none, is_empty, is_in, is_equal, is_true, is_false, is_type, matches]): raise ValueError("no comparison flag specified for verification") if param_compare is None and needs_compare: raise TypeError("'param_compare' cannot be 'None' with specified test flags") is_cmp_typ = isinstance(param_compare, type) or ( isinstance(param_compare, tuple) and param_compare and all(isinstance(_cmp, type) for _cmp in param_compare) ) if is_cmp_typ: # avoid calling 'in' or '__eq__' implementation that could have trouble with 'other' as str type is_str_typ = param_compare in six.string_types or param_compare == six.string_types else: is_str_typ = False if needs_compare and not needs_iterable: # allow 'different' string literals for comparison, otherwise types between value/compare must match exactly # with 'is_type', comparison must be made directly with compare as type instead of with instance type is_str_cmp = isinstance(param, six.string_types) ok_str_cmp = isinstance(param_compare, six.string_types) eq_typ_cmp = type(param) is type(param_compare) is_pattern = matches and isinstance(param_compare, Pattern) if is_type and not (is_str_typ or is_cmp_typ): LOGGER.debug("[param: %s] invalid type compare with [param_compare: %s]", type(param), param_compare) raise TypeError("'param_compare' cannot be of non-type with specified verification flags") if matches and not isinstance(param_compare, (six.string_types, Pattern)): LOGGER.debug("[param_compare: %s] invalid type is not a regex string or pattern", type(param_compare)) raise TypeError("'param_compare' for matching verification must be a string or compile regex pattern") if not is_type and not ((is_str_cmp and ok_str_cmp) or (not is_str_cmp and eq_typ_cmp) or is_pattern): # since 'param' depends on provided input by user, it should be a user-side invalid parameter # only exception is if 'param_compare' is not value-based, then developer combined wrong flags if is_str_typ or is_cmp_typ: LOGGER.debug("[param: %s] invalid value compare with [param_compare: %s]", param, param_compare) raise TypeError("'param_compare' must be value-based for specified verification flags") # when both 'param' and 'param_compare' are values, then the types must match # raise immediately since mismatching param types can make following checks fail uncontrollably LOGGER.debug("[param: %s] != [param_compare: %s]", type(param), type(param_compare)) content = apply_param_content(content, param, param_compare, param_name, with_param, param_content, needs_compare, needs_iterable, is_type, {"is_type": False}) raise_http(http_error, http_kwargs=http_kwargs, detail=msg_on_fail, content=content, content_type=content_type, metadata=metadata) if needs_iterable and (not hasattr(param_compare, "__iter__") or is_str_typ or is_cmp_typ): LOGGER.debug("[param_compare: %s]", param_compare) raise TypeError("'param_compare' must be an iterable of values for specified verification flags") except HTTPException: raise except Exception as exc: content["traceback"] = repr(exc_info()) content["exception"] = repr(exc) raise_http(http_error=HTTPInternalServerError, http_kwargs=http_kwargs, content=content, content_type=content_type, metadata=metadata, detail="Error occurred during parameter verification") # passed this point, input condition flags are valid, evaluate requested parameter combinations fail_conditions = {} fail_verify = False if not_none: fail_conditions.update({"not_none": param is not None}) fail_verify = fail_verify or not fail_conditions["not_none"] if is_none: fail_conditions.update({"is_none": param is None}) fail_verify = fail_verify or not fail_conditions["is_none"] if is_true: fail_conditions.update({"is_true": param is True}) fail_verify = fail_verify or not fail_conditions["is_true"] if is_false: fail_conditions.update({"is_false": param is False}) fail_verify = fail_verify or not fail_conditions["is_false"] if not_empty: fail_conditions.update({"not_empty": hasattr(param, "__len__") and len(param) > 0}) fail_verify = fail_verify or not fail_conditions["not_empty"] if is_empty: fail_conditions.update({"is_empty": hasattr(param, "__len__") and len(param) == 0}) fail_verify = fail_verify or not fail_conditions["is_empty"] if not_in: fail_conditions.update({"not_in": param not in param_compare}) fail_verify = fail_verify or not fail_conditions["not_in"] if is_in: fail_conditions.update({"is_in": param in param_compare}) fail_verify = fail_verify or not fail_conditions["is_in"] if not_equal: fail_conditions.update({"not_equal": param != param_compare}) fail_verify = fail_verify or not fail_conditions["not_equal"] if is_equal: fail_conditions.update({"is_equal": param == param_compare}) fail_verify = fail_verify or not fail_conditions["is_equal"] if is_type: fail_conditions.update({"is_type": isinstance(param, param_compare)}) fail_verify = fail_verify or not fail_conditions["is_type"] if matches: param_compare_regex = param_compare if isinstance(param_compare, six.string_types): param_compare_regex = re.compile(param_compare, re.X) fail_conditions.update({"matches": bool(re.match(param_compare_regex, param))}) fail_verify = fail_verify or not fail_conditions["matches"] if fail_verify: content = apply_param_content(content, param, param_compare, param_name, with_param, param_content, needs_compare, needs_iterable, is_type, fail_conditions) raise_http(http_error, http_kwargs=http_kwargs, detail=msg_on_fail, content=content, content_type=content_type, metadata=metadata)
[docs] def apply_param_content(content, # type: JSON param, # type: Any param_compare, # type: Any param_name, # type: Str with_param, # type: bool param_content, # type: Optional[JSON] needs_compare, # type: bool needs_iterable, # type: bool is_type, # type: bool fail_conditions, # type: JSON ): # type: (...) -> JSON """ Formats and applies the failing parameter conditions and results to returned JSON content according to flags. .. seealso:: :func:`verify_param` """ if with_param: content["param"] = {} content["param"]["conditions"] = fail_conditions if isinstance(param, six.string_types + (int, float, bool, type(None))): # type: ignore content["param"]["value"] = param else: content["param"]["value"] = str(param) if param_name is not None: content["param"]["name"] = str(param_name) if needs_compare and param_compare is not None: if needs_iterable or is_type: param_compare = str if param_compare == six.string_types else param_compare param_compare = getattr(param_compare, "__name__", str(param_compare)) param_compare = "Type[{}]".format(param_compare) if is_type else param_compare if isinstance(param_compare, Pattern): param_compare = param_compare.pattern content["param"]["compare"] = str(param_compare) if isinstance(param_content, dict): content["param"].update(param_content) return content
[docs] def evaluate_call(call, # type: Callable[[], Any] fallback=None, # type: Optional[Callable[[], None]] http_error=HTTPInternalServerError, # type: Type[HTTPError] http_kwargs=None, # type: Optional[ParamsType] msg_on_fail="", # type: Str content=None, # type: Optional[JSON] content_type=CONTENT_TYPE_JSON, # type: Str metadata=None, # type: Optional[JSON] ): # type: (...) -> Any """ Evaluates the specified :paramref:`call` with a wrapped HTTP exception handling. On failure, tries to call. :paramref:`fallback` if specified, and finally raises the specified :paramref:`http_error`. Any potential error generated by :paramref:`fallback` or :paramref:`http_error` themselves are treated as :class:`HTTPInternalServerError`. Exceptions are generated using the standard output method formatted based on specified :paramref:`content_type`. Example: normal call:: try: res = func(args) except Exception as exc: fb_func() raise HTTPExcept(exc.message) wrapped call:: res = evaluate_call(lambda: func(args), fallback=lambda: fb_func(), http_error=HTTPExcept, **kwargs) :param call: function to call, *MUST* be specified as `lambda: <function_call>` :param fallback: function to call (if any) when `call` failed, *MUST* be `lambda: <function_call>` :param http_error: alternative exception to raise on `call` failure :param http_kwargs: additional keyword arguments to pass to `http_error` if called in case of HTTP exception :param msg_on_fail: message details to return in HTTP exception if `call` failed :param content: json formatted additional content to provide in case of exception :param content_type: format in which to return the exception (one of `magpie.common.SUPPORTED_ACCEPT_TYPES`) :param metadata: request metadata to add to the response body. (see: :func:`magpie.api.requests.get_request_info`) :raises http_error: on `call` failure :raises `HTTPInternalServerError`: on `fallback` failure :return: whichever return value `call` might have if no exception occurred """ msg_on_fail = str(msg_on_fail) if isinstance(msg_on_fail, six.string_types) else repr(msg_on_fail) content_repr = repr(content) if content is not None else content if not islambda(call): raise_http(http_error=HTTPInternalServerError, http_kwargs=http_kwargs, metadata=metadata, detail="Input 'call' is not a lambda expression.", content={"call": {"detail": msg_on_fail, "content": content_repr}}, content_type=content_type) # preemptively check fallback to avoid possible call exception without valid recovery if fallback is not None: if not islambda(fallback): raise_http(http_error=HTTPInternalServerError, http_kwargs=http_kwargs, metadata=metadata, detail="Input 'fallback' is not a lambda expression, not attempting 'call'.", content={"call": {"detail": msg_on_fail, "content": content_repr}}, content_type=content_type) try: return call() except Exception as exc: exc_call = {"exception": type(exc).__name__, "error": str(exc), "detail": msg_on_fail, "content": content_repr, "type": content_type} LOGGER.debug("Exception during call evaluation: %s", exc_call, exc_info=exc) try: if fallback is not None: fallback() except Exception as exc: exc_fallback = {"exception": type(exc).__name__, "error": str(exc)} raise_http(http_error=HTTPInternalServerError, http_kwargs=http_kwargs, metadata=metadata, detail="Exception occurred during 'fallback' called after failing 'call' exception.", content={"call": exc_call, "fallback": exc_fallback}, content_type=content_type) raise_http(http_error, detail=msg_on_fail, http_kwargs=http_kwargs, metadata=metadata, content={"call": exc_call}, content_type=content_type)
[docs] def valid_http(http_success=HTTPOk, # type: Union[Type[HTTPSuccessful], Type[HTTPRedirection]] http_kwargs=None, # type: Optional[ParamsType] detail="", # type: Optional[Str] content=None, # type: Optional[JSON] content_type=CONTENT_TYPE_JSON, # type: Optional[Str] metadata=None, # type: Optional[JSON] ): # type: (...) -> Union[HTTPSuccessful, HTTPRedirection] """ Returns successful HTTP with standardized information formatted with content type. (see :func:`raise_http` for HTTP error calls) :param http_success: any derived class from *valid* HTTP codes (<400) (default: `HTTPOk`) :param http_kwargs: additional keyword arguments to pass to `http_success` when called :param detail: additional message information (default: empty) :param content: json formatted content to include :param content_type: format in which to return the exception (one of `magpie.common.SUPPORTED_ACCEPT_TYPES`) :param metadata: request metadata to add to the response body. (see: :func:`magpie.api.requests.get_request_info`) :returns: formatted successful response with additional details and HTTP code """ global RAISE_RECURSIVE_SAFEGUARD_COUNT # pylint: disable=W0603 content = {} if content is None else content detail = repr(detail) if not isinstance(detail, six.string_types) else detail content_type = CONTENT_TYPE_JSON if content_type == CONTENT_TYPE_ANY else content_type http_code, detail, content = validate_params(http_success, [HTTPSuccessful, HTTPRedirection], detail, content, content_type) json_body = format_content_json_str(http_code, detail, content, content_type) resp = generate_response_http_format(http_success, http_kwargs, json_body, content_type=content_type, metadata=metadata) RAISE_RECURSIVE_SAFEGUARD_COUNT = 0 # reset counter for future calls (don't accumulate for different requests) return resp # noqa
[docs] def raise_http(http_error=HTTPInternalServerError, # type: Type[HTTPError] http_kwargs=None, # type: Optional[ParamsType] detail="", # type: Str content=None, # type: Optional[JSON] content_type=CONTENT_TYPE_JSON, # type: Str metadata=None, # type: Optional[JSON] nothrow=False # type: bool ): # type: (...) -> NoReturn """ Raises error HTTP with standardized information formatted with content type. The content contains the corresponding http error code, the provided message as detail and optional specified additional json content (kwarg dict). .. seealso:: :func:`valid_http` for HTTP successful calls :param http_error: any derived class from base `HTTPError` (default: `HTTPInternalServerError`) :param http_kwargs: additional keyword arguments to pass to `http_error` if called in case of HTTP exception :param detail: additional message information (default: empty) :param content: JSON formatted content to include :param content_type: format in which to return the exception (one of `magpie.common.SUPPORTED_ACCEPT_TYPES`) :param metadata: request metadata to add to the response body. (see: :func:`magpie.api.requests.get_request_info`) :param nothrow: returns the error response instead of raising it automatically, but still handles execution errors :raises HTTPError: formatted raised exception with additional details and HTTP code :returns: HTTPError formatted exception with additional details and HTTP code only if `nothrow` is `True` """ # fail-fast if recursion generates too many calls # this would happen only if a major programming error occurred within this function global RAISE_RECURSIVE_SAFEGUARD_MAX # pylint: disable=W0602,W0603 global RAISE_RECURSIVE_SAFEGUARD_COUNT # pylint: disable=W0602,W0603 RAISE_RECURSIVE_SAFEGUARD_COUNT = RAISE_RECURSIVE_SAFEGUARD_COUNT + 1 if RAISE_RECURSIVE_SAFEGUARD_COUNT > RAISE_RECURSIVE_SAFEGUARD_MAX: raise HTTPInternalServerError(detail="Terminated. Too many recursions of `raise_http`") # try dumping content with json format, `HTTPInternalServerError` with caller info if fails. # content is added manually to avoid auto-format and suppression of fields by `HTTPException` content_type = CONTENT_TYPE_JSON if content_type == CONTENT_TYPE_ANY else content_type _, detail, content = validate_params(http_error, HTTPError, detail, content, content_type) json_body = format_content_json_str(http_error.code, detail, content, content_type) resp = generate_response_http_format(http_error, http_kwargs, json_body, content_type=content_type, metadata=metadata) # reset counter for future calls (don't accumulate for different requests) # following raise is the last in the chain since it wasn't triggered by other functions RAISE_RECURSIVE_SAFEGUARD_COUNT = 0 if nothrow: return resp raise resp
[docs] def validate_params(http_class, # type: Type[HTTPException] http_base, # type: Union[Type[HTTPException], Iterable[Type[HTTPException]]] detail, # type: Str content, # type: Optional[JSON] content_type, # type: Str ): # type: (...) -> Tuple[int, Str, JSON] """ Validates parameter types and formats required by :func:`valid_http` and :func:`raise_http`. :param http_class: any derived class from base `HTTPException` to verify :param http_base: any derived sub-class(es) from base `HTTPException` as minimum requirement for `http_class` (ie: 2xx, 4xx, 5xx codes). Can be a single class of an iterable of possible requirements (any). :param detail: additional message information (default: empty) :param content: json formatted content to include :param content_type: format in which to return the exception (one of `magpie.common.SUPPORTED_ACCEPT_TYPES`) :raise `HTTPInternalServerError`: if any parameter is of invalid expected format :returns http_code, detail, content: parameters with corrected and validated format if applicable """ # verify input arguments, raise `HTTPInternalServerError` with caller info if invalid # cannot be done within a try/except because it would always trigger with `raise_http` content = {} if content is None else content detail = repr(detail) if not isinstance(detail, six.string_types) else detail caller = {"content": content, "type": content_type, "detail": detail, "code": 520} # "unknown" code error verify_param(isclass(http_class), param_name="http_class", is_true=True, http_error=HTTPInternalServerError, content_type=CONTENT_TYPE_JSON, content={"caller": caller}, msg_on_fail="Object specified is not a class, class derived from `HTTPException` is expected.") # if `http_class` derives from `http_base` (ex: `HTTPSuccessful` or `HTTPError`) it is of proper requested type # if it derives from `HTTPException`, it *could* be different than base (ex: 2xx instead of 4xx codes) # return 'unknown error' (520) if not of lowest level base `HTTPException`, otherwise use the available code http_base = tuple(http_base if hasattr(http_base, "__iter__") else [http_base]) if issubclass(http_class, http_base): http_code = http_class.code # noqa elif issubclass(http_class, HTTPException): http_code = http_class.code else: http_code = 520 caller["code"] = http_code verify_param(issubclass(http_class, http_base), param_name="http_base", is_true=True, http_error=HTTPInternalServerError, content_type=CONTENT_TYPE_JSON, content={"caller": caller}, msg_on_fail="Invalid 'http_base' derived class specified.") verify_param(content_type, param_name="content_type", param_compare=SUPPORTED_ACCEPT_TYPES, is_in=True, http_error=HTTPInternalServerError, content_type=CONTENT_TYPE_JSON, content={"caller": caller}, msg_on_fail="Invalid 'content_type' specified for exception output.") return http_code, detail, content
[docs] def format_content_json_str(http_code, detail, content, content_type): # type: (int, Str, JSON, Str) -> Str """ Inserts the code, details, content and type within the body using json format. Includes also any other specified json formatted content in the body. Returns the whole json body as a single string for output. :raise `HTTPInternalServerError`: if parsing of the json content failed. :returns: formatted JSON content as string with added HTTP code and details. """ json_body = "" try: content["code"] = http_code content["detail"] = detail content["type"] = content_type json_body = json.dumps(content) except Exception as exc: # pylint: disable=W0703 msg = "Dumping json content '{!s}' resulted in exception '{!r}'.".format(content, exc) raise_http(http_error=HTTPInternalServerError, detail=msg, content_type=CONTENT_TYPE_JSON, content={"traceback": repr(exc_info()), "exception": repr(exc), "caller": {"content": repr(content), # raw string to avoid recursive json.dumps error "detail": detail, "code": http_code, "type": content_type}}) return json_body
[docs] def rewrite_content_type(content, content_type): # type: (Union[Str, JSON], Str) -> Tuple[Str, Optional[JSON]] """ Attempts to rewrite the ``type`` field inserted by various functions such as: - :func:`format_content_json_str` - :func:`raise_http` - :func:`valid_http` By applying the new value provided by :paramref:`content_type`. :returns: Content with rewritten "type" (if possible) and converted to string directly insertable to a response body. Also provides the converted JSON body if applicable (original content was literal JSON or JSON-like string). """ json_content = None if isinstance(content, six.string_types): try: content = json.loads(content) json_content = content except (TypeError, json.decoder.JSONDecodeError): pass if isinstance(content, (list, dict)): if "type" in content: content["type"] = content_type json_content = content content = json.dumps(content) return content, json_content
[docs] def generate_response_http_format(http_class, http_kwargs, content, content_type=CONTENT_TYPE_PLAIN, metadata=None): # type: (Type[HTTPException], Optional[ParamsType], JSON, Optional[Str], Optional[JSON]) -> HTTPException """ Formats the HTTP response content according to desired ``content_type`` using provided HTTP code and content. :param http_class: `HTTPException` derived class to use for output (code, generic title/explanation, etc.) :param http_kwargs: additional keyword arguments to pass to `http_class` when called :param content: formatted JSON content or literal string content providing additional details for the response :param content_type: one of `magpie.common.SUPPORTED_ACCEPT_TYPES` (default: `magpie.common.CONTENT_TYPE_PLAIN`) :param metadata: request metadata to add to the response body. (see: :func:`magpie.api.requests.get_request_info`) :return: `http_class` instance with requested information and content type if creation succeeds :raises: `HTTPInternalServerError` instance details about requested information and content type if creation fails """ # content body is added manually to avoid auto-format and suppression of fields by `HTTPException` content, json_content = rewrite_content_type(content, content_type) if isinstance(json_content, dict) and isinstance(metadata, dict): # ensure that original JSON content has priority in fields definition over metadata # preserve original JSON field ordering, as best as possible json_content.update({k: v for k, v in metadata.items() if k not in json_content}) content, json_content = rewrite_content_type(json_content, content_type) content = str(content) if not isinstance(content, six.string_types) else content # adjust additional keyword arguments and try building the http response class with them http_kwargs = {} if http_kwargs is None else http_kwargs http_headers = http_kwargs.get("headers", {}) # omit content-type and related headers that we override for header in dict(http_headers): if header.lower().startswith("content-"): http_headers.pop(header, None) try: # Pass down Location if it is provided and should be given as input parameter for this HTTP class. # Omitting this step would inject a (possibly extra) empty Location that defaults to the current application. # When resolving HTTP redirects, injecting this extra Location when the requested one is not the current # application will lead to redirection failures because all locations are appended in the header as CSV list. if issubclass(http_class, HTTPRedirection): location = get_header("Location", http_headers, pop=True) if location and "location" not in http_kwargs: http_kwargs["location"] = location # directly output json if content_type == CONTENT_TYPE_JSON: content_type = "{}; charset=UTF-8".format(CONTENT_TYPE_JSON) http_response = http_class(body=content, content_type=content_type, **http_kwargs) # otherwise json is contained within the html <body> section elif content_type == CONTENT_TYPE_HTML: if http_class is HTTPOk: http_class.explanation = "Operation successful." if not http_class.explanation: http_class.explanation = http_class.title # some don't have any defined # add preformat <pre> section to output as is within the <body> section html_status = "Exception" if http_class.code >= 400 else "Response" html_header = "{}<br><h2>{} Details</h2>".format(http_class.explanation, html_status) html_template = "<pre style='word-wrap: break-word; white-space: pre-wrap;'>{}</pre>" content_type = "{}; charset=UTF-8".format(CONTENT_TYPE_HTML) if json_content: html_body = html_template.format(json.dumps(json_content, indent=True, ensure_ascii=False)) else: html_body = html_template.format(content) html_body = html_header + html_body http_response = http_class(body_template=html_body, content_type=content_type, **http_kwargs) elif content_type in [CONTENT_TYPE_APP_XML, CONTENT_TYPE_TXT_XML]: xml_body = dicttoxml(json_content, custom_root="response") http_response = http_class(body=xml_body, content_type=CONTENT_TYPE_TXT_XML, **http_kwargs) # default back to plain text else: http_response = http_class(body=content, content_type=CONTENT_TYPE_PLAIN, **http_kwargs) return http_response except Exception as exc: # pylint: disable=W0703 raise_http(http_error=HTTPInternalServerError, detail="Failed to build HTTP response", content={"traceback": repr(exc_info()), "exception": repr(exc), "caller": {"http_kwargs": repr(http_kwargs), "http_class": repr(http_class), "content_type": str(content_type)}})