Source code for

import re
from typing import Any, Callable, Dict, Optional, Union

from globus_sdk.response import GlobusHTTPResponse, IterableResponse

[docs]class IterableGCSResponse(IterableResponse): """ Response class for non-paged list oriented resources. Allows top level fields to be accessed normally via standard item access, and also provides a convenient way to iterate over the sub-item list in the ``data`` key: >>> print("Path:", r["path"]) >>> # Equivalent to: for item in r["data"] >>> for item in r: >>> print(item["name"], item["type"]) """ default_iter_key = "data"
[docs]class UnpackingGCSResponse(GlobusHTTPResponse): """ An "unpacking" response looks for a "data" array in the response data, which is expected to have dict elements. The "data" is traversed until the first matching object is found, and this is presented as the ``data`` property of the response. The full response data is available as ``full_data``. If the expected datatype is not found in the array, or the array is missing, the ``data`` will be the full response data (identical to ``full_data``). :param match: Either a string containing a DATA_TYPE prefix, or an arbitrary callable which does the matching :type match: str or callable """ def _default_unpacking_match(self, spec: str) -> Callable[[Dict[str, Any]], bool]: if not re.fullmatch(r"\w+", spec): raise ValueError("Invalid UnpackingGCSResponse specification.") def match_func(data: Dict[str, Any]) -> bool: if not ("DATA_TYPE" in data and isinstance(data["DATA_TYPE"], str)): return False if "#" not in data["DATA_TYPE"]: return False name, _version = data["DATA_TYPE"].split("#", 1) return name == spec return match_func def __init__( self, response: GlobusHTTPResponse, match: Union[str, Callable[[Dict[str, Any]], bool]], ): super().__init__(response) if callable(match): self._match_func = match else: self._match_func = self._default_unpacking_match(match) self._unpacked_data: Optional[Dict[str, Any]] = None self._did_unpack = False @property def full_data(self) -> Any: """ The full, parsed JSON response data. ``None`` if the data cannot be parsed as JSON. """ return self._parsed_json def _unpack(self) -> Optional[Dict[str, Any]]: """ Unpack the response from the `"data"` array, returning the first match found. If no matches are founds, or the data is the wrong shape, return None. """ if isinstance(self._parsed_json, dict) and isinstance( self._parsed_json.get("data"), list ): for item in self._parsed_json["data"]: if isinstance(item, dict) and self._match_func(item): return item return None @property def data(self) -> Any: # only do the unpacking operation once, as it may be expensive on large payloads if not self._did_unpack: self._unpacked_data = self._unpack() self._did_unpack = True if self._unpacked_data is not None: return self._unpacked_data return self._parsed_json