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":
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
