from __future__ import annotations
import enum
import logging
import typing as t
import requests
from globus_sdk import _guards
from .base import GlobusError
from .err_info import ErrorInfoContainer
from .warnings import warn_deprecated
log = logging.getLogger(__name__)
_CACHE_SENTINEL = object()
class _ErrorFormat(enum.Enum):
undefined = enum.auto()
jsonapi = enum.auto()
type_zero = enum.auto()
[docs]
class GlobusAPIError(GlobusError):
"""
Wraps errors returned by a REST API.
:ivar int http_status: HTTP status code
:ivar str code: Error code from the API or "Error" for unclassified errors
:ivar str request_id: The 'request_id' included in the error data, if any.
:ivar list[str] messages: A list of error messages, extracted from the response
data. If the data cannot be parsed or does not contain any clear message fields,
this list may be empty.
:ivar list[GlobusSubError] errors: A list of sub-error documents, as would be
presented by JSON:API APIs and similar interfaces.
"""
MESSAGE_FIELDS = ["message", "detail", "title"]
RECOGNIZED_AUTHZ_SCHEMES = ["bearer", "basic", "globus-goauthtoken"]
def __init__(self, r: requests.Response, *args: t.Any, **kwargs: t.Any):
self._cached_raw_json: t.Any = _CACHE_SENTINEL
self.http_status = r.status_code
# defaults, may be rewritten during parsing
self.code: str | None = "Error"
self.request_id: str | None = None
self.messages: list[str] = []
self.errors: list[ErrorSubdocument] = []
self._info: ErrorInfoContainer | None = None
self._underlying_response = r
self._parse_response()
super().__init__(*self._get_args())
@property
def message(self) -> str | None:
"""
An error message from the API.
If there are multiple messages available, this will contain all messages joined
with semicolons. If there is no message available, this will be ``None``.
"""
if self.messages:
return "; ".join(self.messages)
return None
@message.setter
def message(self, value: str) -> None:
warn_deprecated(
"Setting a message on GlobusAPIError objects is deprecated. "
"This overwrites any parsed messages. Append to 'messages' instead."
)
self.messages = [value]
@property
def http_reason(self) -> str:
"""
The HTTP reason string from the response.
This is the part of the status line after the status code, and typically is a
string description of the status. If the status line is
``HTTP/1.1 404 Not Found``, then this is the string ``"Not Found"``.
"""
return self._underlying_response.reason
@property
def headers(self) -> t.Mapping[str, str]:
"""
The HTTP response headers as a case-insensitive mapping.
For example, ``headers["Content-Length"]`` and ``headers["content-length"]`` are
treated as equivalent.
"""
return self._underlying_response.headers
@property
def content_type(self) -> str | None:
return self.headers.get("Content-Type")
def _get_mimetype(self, content_type: str) -> str:
return content_type.split(";")[0].strip()
def _jsonapi_mimetype(self) -> bool:
if self.content_type is None:
return False
return self._get_mimetype(self.content_type) == "application/vnd.api+json"
def _json_mimetype(self) -> bool:
if self.content_type is None:
return False
mimetype = self._get_mimetype(self.content_type)
if mimetype == "application/json":
return True
if mimetype.startswith("application/") and mimetype.endswith("+json"):
return True
return False
@property
def raw_json(self) -> dict[str, t.Any] | None:
"""
Get the verbatim error message received from a Globus API, interpreted
as JSON data
If the body cannot be loaded as JSON, this is None
"""
if self._cached_raw_json == _CACHE_SENTINEL:
self._cached_raw_json = None
if self._json_mimetype():
try:
# technically, this could be a non-dict JSON type, like a list or
# string but in those cases the user can just cast -- the "normal"
# case is a dict
self._cached_raw_json = self._underlying_response.json()
except ValueError:
log.error(
"Error body could not be JSON decoded! "
"This means the Content-Type is wrong, or the "
"body is malformed!"
)
return t.cast("dict[str, t.Any]", self._cached_raw_json)
@property
def _dict_data(self) -> dict[str, t.Any]:
"""
A "type asserting" wrapper over raw_json which errors if the type is not dict.
"""
if not isinstance(self.raw_json, dict):
raise ValueError("cannot use _dict_data when data is non-dict type")
return self.raw_json
@property
def text(self) -> str:
"""
Get the verbatim error message received from a Globus API as a *string*
"""
return self._underlying_response.text
@property
def raw_text(self) -> str:
"""
Deprecated alias of the ``text`` property.
"""
warn_deprecated(
"The 'raw_text' property of GlobusAPIError objects is deprecated. "
"Use the 'text' property instead."
)
return self.text
@property
def binary_content(self) -> bytes:
"""
The error message received from a Globus API in bytes.
"""
return self._underlying_response.content
@property
def info(self) -> ErrorInfoContainer:
"""
An ``ErrorInfoContainer`` with parsed error data. The ``info`` of an error is
guaranteed to be present, but all of its contents may be falsey if the error
could not be parsed.
"""
if self._info is None:
json_data = self.raw_json if isinstance(self.raw_json, dict) else None
self._info = ErrorInfoContainer(json_data)
return self._info
def _get_request_authorization_scheme(self) -> str | None:
try:
authz_h = self._underlying_response.request.headers["Authorization"]
authz_scheme = authz_h.split()[0]
if authz_scheme.lower() in self.RECOGNIZED_AUTHZ_SCHEMES:
return authz_scheme
except (IndexError, KeyError):
pass
return None
def _get_args(self) -> list[t.Any]:
"""
Get arguments to pass to the Exception base class. These args are
displayed in stack traces.
"""
args = [
self._underlying_response.request.method,
self._underlying_response.url,
self._get_request_authorization_scheme(),
self.http_status,
self.code,
# if the message is "", try using response reason
# for details on these, and some examples, see
# https://datatracker.ietf.org/doc/html/rfc7231#section-6.1
self.message or self._underlying_response.reason,
]
if self.request_id:
args.append(self.request_id)
return args
def _parse_response(self) -> bool:
"""
This is an intermediate step between 'raw_json' (loading bare JSON data)
and the "real" parsing methods.
In order to better support subclassing with short-circuiting behaviors if
parsing goes awry, all of the parsing methods return True if parsing should
continue or False if it was aborted.
_parse_response() pulls the JSON body and does the following:
- on non-dict JSON data, log and abort early. Don't error since this
is already part of exception construction, just stop parsing.
- Attempt to detect the error format in use, then dispatch to the relevant
subparser: JSON:API, Type Zero, or Undefined
- if subparsing succeeded, call the `_post_parse_hook` hook to allow
subclasses to trivially add more computed attributes. This could also be
done by altering `__init__` but it's nicer to have a dedicated hook because
it is guaranteed to only be called if the rest of parsing succeeded
"""
if self.raw_json is None:
log.debug("Error body was not parsed as JSON")
return False
if not isinstance(self.raw_json, dict):
log.warning( # type: ignore[unreachable]
"Error body could not be parsed as JSON because it was not a dict"
)
return False
error_format = self._detect_error_format()
subparse_result: bool
if error_format == _ErrorFormat.jsonapi:
subparse_result = self._parse_jsonapi_error_format()
elif error_format == _ErrorFormat.type_zero:
subparse_result = self._parse_type_zero_error_format()
else:
subparse_result = self._parse_undefined_error_format()
if not subparse_result:
return False
return self._post_parse_hook()
def _post_parse_hook(self) -> bool:
"""
An internal callback for extra customizations after fully successful parsing.
By default, does nothing.
"""
return True
def _detect_error_format(self) -> _ErrorFormat:
# if the JSON:API mimetype was used, inspect the data to make sure it is
# well-formed
if self._jsonapi_mimetype():
errors = self._dict_data.get("errors")
if not _guards.is_list_of(errors, dict):
return _ErrorFormat.undefined
elif len(errors) < 1:
return _ErrorFormat.undefined
elif not all(isinstance(error_doc, dict) for error_doc in errors):
return _ErrorFormat.undefined
# only use 'jsonapi' if everything checked out
return _ErrorFormat.jsonapi
# now evaluate attributes for Type Zero errors under the same paradigm
# check each attribute and only return 'type_zero' if nothing failed
if not isinstance(self._dict_data.get("code"), str):
return _ErrorFormat.undefined
elif not isinstance(self._dict_data.get("message"), str):
return _ErrorFormat.undefined
# request_id is not required, but must be a string if present
elif "request_id" in self._dict_data and not isinstance(
self._dict_data["request_id"], str
):
return _ErrorFormat.undefined
return _ErrorFormat.type_zero
def _parse_jsonapi_error_format(self) -> bool:
"""
Parsing a JSON:API Error
This is only called after the field type for 'errors' has been checked.
However, the nested/underlying fields will not have been checked yet.
"""
self.errors = [ErrorSubdocument(e) for e in self._dict_data["errors"]]
self.code = self._extract_code_from_error_array(self.errors)
self.messages = self._extract_messages_from_error_array(self.errors)
return True
def _parse_type_zero_error_format(self) -> bool:
"""
Parsing a Type Zero Error
This is only called after Type Zero has been detected. Therefore, we already
have assurances about the values in 'code' and 'message'.
Note that 'request_id' could be absent but *must* be a string if present.
"""
self.code = self._dict_data["code"]
self.messages = [self._dict_data["message"]]
self.request_id = self._dict_data.get("request_id")
if _guards.is_list_of(self._dict_data.get("errors"), dict):
raw_errors = self._dict_data["errors"]
else:
raw_errors = [self._dict_data]
self.errors = [
ErrorSubdocument(e, message_fields=self.MESSAGE_FIELDS) for e in raw_errors
]
return True
def _parse_undefined_error_format(self) -> bool:
"""
Undefined Parsing: best effort support for unknown data shapes
This is also a great place for custom parsing to hook in for different APIs
if we know that there's an unusual format in use
"""
# attempt to pull out errors if possible and valid
if _guards.is_list_of(self._dict_data.get("errors"), dict):
raw_errors = self._dict_data["errors"]
# if no 'errors' were found, or 'errors' is invalid, then
# 'errors' should be set to contain the root document
else:
raw_errors = [self._dict_data]
self.errors = [
ErrorSubdocument(e, message_fields=self.MESSAGE_FIELDS) for e in raw_errors
]
# use 'code' if present and correct type
if isinstance(self._dict_data.get("code"), str):
self.code = t.cast(str, self._dict_data["code"])
# otherwise, pull 'code' from the sub-errors array
elif self.errors:
# in undefined parse cases, the code will be left as `"Error"` for
# a higher degree of backwards compatibility
maybe_code = self._extract_code_from_error_array(self.errors)
if maybe_code is not None:
self.code = maybe_code
# either there is an array of subdocument errors or there is not,
# in which case we load only from the root doc
self.messages = self._extract_messages_from_error_array(
self.errors
or [ErrorSubdocument(self._dict_data, message_fields=self.MESSAGE_FIELDS)]
)
return True
def _extract_messages_from_error_array(
self, errors: list[ErrorSubdocument]
) -> list[str]:
"""
Extract 'messages' from an array of errors (JSON:API or otherwise)
Each subdocument *may* define its messages, so this is the aggregate of messages
from those documents which had messages.
Note that subdocuments may be instructed about their `message_fields` by the
error class and parsing path which they take. Therefore, this may be extracting
`"message"`, `"detail"`, or `"title"` in the base implementation and other
fields if a subclass customizes this further.
"""
ret: list[str] = []
for doc in errors:
if doc.message is not None:
ret.append(doc.message)
return ret
def _extract_code_from_error_array(
self, errors: list[ErrorSubdocument]
) -> str | None:
"""
Extract a 'code' field from an array of errors (JSON:API or otherwise)
This is done by checking if each error document has the same 'code'
"""
codes: set[str] = set()
for error in errors:
if error.code is not None:
codes.add(error.code)
if len(codes) == 1:
return codes.pop()
return None
[docs]
class ErrorSubdocument:
"""
Error subdocuments as returned by Globus APIs.
:ivar dict raw: The unparsed error subdocument
"""
# the default set of fields to use for message extraction, in order
# selected to match the fields defined by JSON:API by default
DEFAULT_MESSAGE_FIELDS: tuple[str, ...] = ("detail", "title")
def __init__(
self, data: dict[str, t.Any], *, message_fields: t.Sequence[str] | None = None
) -> None:
self.raw = data
self._message_fields: tuple[str, ...]
if message_fields is not None:
self._message_fields = tuple(message_fields)
else:
self._message_fields = self.DEFAULT_MESSAGE_FIELDS
@property
def message(self) -> str | None:
"""
The 'message' string of this subdocument, derived from its data based on the
parsing context.
May be `None` if no message is defined.
"""
return _extract_message_from_dict(self.raw, self._message_fields)
@property
def code(self) -> str | None:
"""
The 'code' string of this subdocument, derived from its data based on the
parsing context.
May be `None` if no code is defined.
"""
if isinstance(self.raw.get("code"), str):
return t.cast(str, self.raw["code"])
return None
[docs]
def get(self, key: str, default: t.Any = None) -> t.Any | None:
"""
A dict-like getter for the raw data.
:param key: The string key to use for lookup
:param default: The default value to use
"""
return self.raw.get(key, default)
def _extract_message_from_dict(
data: dict[str, t.Any], message_fields: tuple[str, ...]
) -> str | None:
"""
Extract a single message string from a dict if one is present.
"""
for f in message_fields:
if isinstance(data.get(f), str):
return t.cast(str, data[f])
return None