Source code for globus_sdk.response

from __future__ import annotations

import json
import logging
import typing as t

from requests import Response

from globus_sdk import _guards

log = logging.getLogger(__name__)

    import globus_sdk

[docs] class GlobusHTTPResponse: """ Response object that wraps an HTTP response from the underlying HTTP library. If the response is JSON, the parsed data will be available in ``data``, otherwise ``data`` will be ``None`` and ``text`` should be used instead. The most common response data is a JSON dictionary. To make handling this type of response as seamless as possible, the ``GlobusHTTPResponse`` object implements the immutable mapping protocol for dict-style access. This is just an alias for access to the underlying data. If the response data is not a dictionary or list, item access will raise ``TypeError``. >>> print("Response ID": r["id"]) # alias for["id"] :ivar client: The client instance which made the request """ def __init__( self, response: Response | GlobusHTTPResponse, client: globus_sdk.BaseClient | None = None, ): # init on a GlobusHTTPResponse: we are wrapping this data # the _response is None if isinstance(response, GlobusHTTPResponse): if client is not None: raise ValueError("Redundant client with wrapped response") self._wrapped: GlobusHTTPResponse | None = response self._response: Response | None = None self.client: globus_sdk.BaseClient = self._wrapped.client # copy parsed JSON data off of '_wrapped' self._parsed_json: t.Any = self._wrapped._parsed_json # init on a Response object, this is the "normal" case # _wrapped is None else: if client is None: raise ValueError("Missing client with normal response") self._wrapped = None self._response = response self.client = client # JSON decoding may raise a ValueError due to an invalid JSON # document. In the case of trying to fetch the "data" on an HTTP # response, this means we didn't get a JSON response. # store this as None, as in "no data" # # if the caller *really* wants the raw body of the response, they can # always use `text` try: self._parsed_json = self._response.json() except ValueError: log.warning("response data did not parse as JSON, data=None") self._parsed_json = None @property def _raw_response(self) -> Response: # this is an internal property which traverses any series of wrapped responses # until reaching a requests response object if self._response is not None: return self._response elif self._wrapped is not None: return self._wrapped._raw_response else: # unreachable # pragma: no cover raise ValueError("could not find an inner response object") @property def http_status(self) -> int: """The HTTP response status, as an integer.""" return self._raw_response.status_code @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 200 OK``, then this is the string ``"OK"``. """ return self._raw_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._raw_response.headers @property def content_type(self) -> str | None: return self.headers.get("Content-Type") @property def text(self) -> str: """The raw response data as a string.""" return self._raw_response.text @property def binary_content(self) -> bytes: """ The raw response data in bytes. """ return self._raw_response.content @property def data(self) -> t.Any: return self._parsed_json
[docs] def get(self, key: str, default: t.Any = None) -> t.Any: """ ``get`` is just an alias for ``data.get(key, default)``, but with the added checks that if ``data`` is ``None`` or a list, it returns the default. :param key: The string key to lookup in the response if it is a dict :param default: The default value to be used if the data is null or a list """ if _guards.is_optional(, list): return default # NB: `default` is provided as a positional because the native dict type # doesn't recognize a keyword argument `default` return, default)
def __str__(self) -> str: """The default __str__ for a response assumes that the data is valid JSON-dump-able.""" if is not None: return json.dumps(, indent=2, separators=(",", ": ")) return self.text def __repr__(self) -> str: return f"{self.__class__.__name__}({self.text})" def __getitem__(self, key: str | int | slice) -> t.Any: # force evaluation of the data property outside of the upcoming # try-catch so that we don't accidentally catch TypeErrors thrown # during the getter function itself data = try: return data[key] except TypeError as err: log.error( f"Can't index into responses with underlying data of type {type(data)}" ) # re-raise with an altered message and error type -- the issue is that # whatever data is in the response doesn't support indexing (e.g. a response # that is just an integer, parsed as json) # # "type" is ambiguous, but we don't know if it's the fault of the # class at large, or just a particular call's `data` property raise ValueError( "This type of response data does not support indexing." ) from err def __contains__(self, item: t.Any) -> bool: """ ``x in response`` is an alias for ``x in`` """ if is None: return False return item in def __bool__(self) -> bool: """ ``bool(response)`` is an alias for ``bool(`` """ return bool(
[docs] class IterableResponse(GlobusHTTPResponse): """This response class adds an __iter__ method on an 'iter_key' variable. The assumption is that iter produces dicts or dict-like mappings.""" default_iter_key: t.ClassVar[str] iter_key: str def __init__( self, response: Response | GlobusHTTPResponse, client: globus_sdk.BaseClient | None = None, *, iter_key: str | None = None, ) -> None: if not hasattr(self, "default_iter_key"): raise TypeError( "Cannot instantiate an iterable response from a class " "which does not define a default iteration key." ) iter_key = iter_key if iter_key is not None else self.default_iter_key self.iter_key = iter_key super().__init__(response, client) def __iter__(self) -> t.Iterator[t.Mapping[t.Any, t.Any]]: if not isinstance(, dict): raise TypeError( "Cannot iterate on IterableResponse data when " f"type is '{type(}'" ) return iter([self.iter_key])
[docs] class ArrayResponse(GlobusHTTPResponse): """This response class adds an ``__iter__`` method which assumes that the top-level data of the response is a JSON array.""" def __iter__(self) -> t.Iterator[t.Any]: if not isinstance(, list): raise TypeError( "Cannot iterate on ArrayResponse data when " f"type is '{type(}'" ) return iter( def __len__(self) -> int: """ ``len(response)`` is an alias for ``len(`` """ if not isinstance(, raise TypeError( "Cannot take len() on ArrayResponse data when " f"type is '{type(}'" ) return len(