Base Client

def __init__(
    self,
    auth_info: Optional[Dict[str, Any]] = None,
    api_info: Optional[Dict[str, Any]] = None,
    user: Optional[str] = None,
    token: Optional[str] = None,
    token_getter: Optional[Callable[[], Tuple[str, int]]] = None,
    prune: bool = False,
    no_verify: bool = False,
    proxy_info: Optional[Dict[str, Any]] = None,
):
    """
        Args:
            auth_info: When specified, a service account or a web auth JSON dict.
            api_info: When specified, a JSON dict containing the description of your
                api. Defaults to LumApps API.
            user: Email of user on behalf of whom to authenticate using domain-wide
                delegation.
            token: A bearer access token.
            token_getter: A bearer access token getter function.
            prune: Whether or not to use FILTERS to prune LumApps API responses.
            no_verify: Disables SSL verification.
            proxy_info: When specified, a JSON dict with proxy parameters.
    """
    self._token_expiry = 0
    self.no_verify = no_verify
    self.proxy_info = proxy_info
    self.prune = prune
    self._auth_info = auth_info
    self._token = None
    self._endpoints = None
    self._client = None
    self._headers: dict = {}
    if api_info is None:
        api_info = {}
    api_info.setdefault("name", LUMAPPS_NAME)
    api_info.setdefault("version", LUMAPPS_VERSION)
    api_info.setdefault("base_url", LUMAPPS_BASE_URL)
    api_info.setdefault("scopes", LUMAPPS_SCOPE)
    self.api_info = api_info
    api_name = api_info["name"]
    self._scope = " ".join(api_info["scopes"])
    api_ver = api_info["version"]
    prefix = "" if api_name in GOOGLE_APIS else "/_ah/api"
    self._api_url = f"{prefix}/{api_name}/{api_ver}"
    self._discovery_url = (
        f"{self.base_url}{prefix}/discovery/v1/apis/{api_name}/{api_ver}/rest"
    )
    self.token_getter = token_getter
    self.user = user
    self.token = token
    self.cursor = None

def get_call(
    self, *name_parts, **params
) -> Union[Dict[str, Any], List[Dict[str, Any]], None]:
    """Generic function to call a lumapps endpoint

        Args:
            *name_parts: Endpoint, eg user/get or "user", "get"
            **params: Parameters of the call

        Returns:
            Object or objects returned by the endpoint call.

        Example:
            List feedtypes in LumApps:
            -> GET https://.../_ah/api/lumsites/v1/feedtype/list

            With this endpoint:

                >>> feedtypes = get_call("feedtype/list")
                >>> print(feedtypes)
    """
    name_parts = _parse_endpoint_parts(name_parts)
    items: List[dict] = []
    self.cursor = cursor = params.pop("cursor", None)
    body = self._pop_body(params)
    while True:
        if cursor:
            if body is not None:
                body["cursor"] = cursor
            else:
                params["cursor"] = cursor
        response = self._call(name_parts, params, body)
        if response is None:
            return None

        more = response.get("more")
        response_items = response.get("items")
        if more:
            if response_items:
                self.cursor = cursor = response["cursor"]
                items.extend(response_items)
            else:
                # No results but a more field set to true ...
                # ie, the api return something wrong
                self.cursor = cursor = None
                return self._prune(name_parts, items)
        else:
            # No more result to get
            if response_items:
                self.cursor = cursor = None
                items.extend(response_items)
                return self._prune(name_parts, items)
            else:
                # No results, return
                self.cursor = cursor = None
                # special case of
                return [] if more is False else self._prune(name_parts, response)

def get_new_client_as(
    self, user_email: str, customer_id: Optional[str] = None
) -> "BaseClient":
    """ Get a new BaseClient using an authorized client account by obtaining a
        token using the user/getToken endpoint.

        Args:
            user_email (str): User you want to authenticate on behalf of
            customer_id (str): Id of the LumApps customer the user belong to

        Returns:
            BaseClient: A new instance of the BaseClient correctly authenticated.
    """
    client = BaseClient(
        auth_info=self._auth_info,
        api_info=self.api_info,
        no_verify=self.no_verify,
        proxy_info=self.proxy_info,
        prune=self.prune,
    )
    token_infos: Any = client.get_call(
        "user/getToken", customerId=customer_id, email=user_email
    )
    token = token_infos["accessToken"]
    return BaseClient(
        api_info=self.api_info,
        token=token,
        user=user_email,
        no_verify=self.no_verify,
        proxy_info=self.proxy_info,
        prune=self.prune,
    )

def get_new_client_as_using_dwd(self, user_email: str) -> "BaseClient":
    """ Get a new BaseClient using domain-wide delegation """
    return BaseClient(
        auth_info=self._auth_info,
        api_info=self.api_info,
        user=user_email,
        no_verify=self.no_verify,
        proxy_info=self.proxy_info,
        prune=self.prune,
    )

def iter_call(
    self, *name_parts, **params
) -> Generator[
    Union[Dict[str, Any], List[Dict[str, Any]]],
    Union[Dict[str, Any], List[Dict[str, Any]]],
    None,
]:
    """
        Args:
            *name_parts: Endpoint, eg user/get or "user", "get"
            **params: Parameters of the call

        Yields:
            Objects returned by the endpoint call


        Example:
            List feedtypes in LumApps:
            -> GET https://.../_ah/api/lumsites/v1/feedtype/list

            With this endpoint:

                >>> feedtypes = iter_call("feedtype/list")
                >>> for feedtype in feedtypes: print(feedtype)
    """
    name_parts = _parse_endpoint_parts(name_parts)
    self.cursor = cursor = params.pop("cursor", None)
    body = self._pop_body(params)
    while True:
        if cursor:
            if body is not None:
                body["cursor"] = cursor
            else:
                params["cursor"] = cursor

        response = self._call(name_parts, params, body)
        more = response.get("more")
        items = response.get("items")

        if more:
            if items:
                # Yield the results and continue the loop
                self.cursor = cursor = response["cursor"]
                for item in items:
                    yield self._prune(name_parts, item)
            else:
                # No results but a more field set to true ...
                # ie, the api return something wrong
                self.cursor = cursor = None
                return
        else:
            # No more result to get
            if items:
                # Yield the last results and then return
                self.cursor = cursor = None
                for item in items:
                    yield self._prune(name_parts, item)
                else:
                    return
            else:
                # No results, return
                self.cursor = cursor = None
                return