from __future__ import annotations
import logging
import typing as t
import requests
from .base import GlobusError
from .err_info import ErrorInfoContainer
from .warnings import warn_deprecated
log = logging.getLogger(__name__)
[docs]class GlobusAPIError(GlobusError):
"""
Wraps errors returned by a REST API.
:ivar http_status: HTTP status code (int)
:ivar code: Error code from the API (str),
or "Error" for unclassified errors
:ivar message: Error message from the API. In general, this will be more
useful to developers, but there may be cases where it's
suitable for display to end users.
"""
MESSAGE_FIELDS = ["message", "detail"]
RECOGNIZED_AUTHZ_SCHEMES = ["bearer", "basic", "globus-goauthtoken"]
def __init__(self, r: requests.Response, *args: t.Any, **kwargs: t.Any):
self.http_status = r.status_code
# defaults, may be rewritten during parsing
self.code = "Error"
self.message = r.text
self._info: ErrorInfoContainer | None = None
self._underlying_response = r
self._parse_response()
super().__init__(*self._get_args())
@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 _json_content_type(self) -> bool:
r = self._underlying_response
return "Content-Type" in r.headers and (
"application/json" in r.headers["Content-Type"]
)
@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
"""
r = self._underlying_response
if not self._json_content_type():
return None
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
return t.cast(t.Dict[str, t.Any], r.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 None
@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:
rawjson = self.raw_json
json_data = rawjson if isinstance(rawjson, 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.
"""
return [
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,
]
def _parse_response(self) -> None:
"""
This is an intermediate step between 'raw_json' (loading bare JSON data)
and the "real" parsing method, '_load_from_json'
_parse_response() pulls the JSON body and does the following:
- if the data is not a dict, ensure _load_from_json is not called (so it only
gets called on dict data)
- if the data contains an 'errors' array, pull out the first error document and
pass that to _load_from_json()
- log a warning on non-dict JSON data
"""
json_data = self.raw_json
if json_data is None:
log.debug("Error body was not parsed as JSON")
return
if not isinstance(json_data, dict):
log.warning( # type: ignore[unreachable]
"Error body could not be parsed as JSON because it was not a dict"
)
return
# if there appears to be a list of errors in the response data, grab the
# first error from that list for parsing
# this is only done if we determine that
# - 'errors' is present and is a non-empty list
# - 'errors[0]' is a dict
#
# this gracefully handles other uses of the key 'errors', e.g. as an int:
# {"message": "foo", "errors": 6}
if (
isinstance(json_data.get("errors"), list)
and len(json_data["errors"]) > 0
and isinstance(json_data["errors"][0], dict)
):
# log a warning only when there is more than one error in the list
# if an API sends back an error of the form
# {"errors": [{"foo": "bar"}]}
# then the envelope doesn't matter and there's only one error to parse
if len(json_data["errors"]) != 1:
log.warning(
"Doing JSON load of error response with multiple "
"errors. Exception data will only include the "
"first error, but there are really %d errors",
len(json_data["errors"]),
)
# try to grab the first error in the list, but also check
# if it isn't a dict
json_data = json_data["errors"][0]
self._load_from_json(json_data)
def _load_from_json(self, data: dict[str, t.Any]) -> None:
# rewrite 'code' if present and correct type
if isinstance(data.get("code"), str):
self.code = data["code"]
for f in self.MESSAGE_FIELDS:
if isinstance(data.get(f), str):
log.debug("Loaded message from '%s' field", f)
self.message = data[f]
break
else:
log.debug("No message found in parsed error body")