Method api_page_iter

bbot/modules/base.py:1359–1420

An asynchronous generator function for iterating through paginated API data. This function continuously makes requests to a specified API URL, incrementing the page number or applying a custom pagination function, and yields the received data one page at a time. It

(self, url, page_size=100, _json=True, next_key=None, iter_key=None, **requests_kwargs)

        return getattr(r, "is_success", False) or getattr(r, "status_code", 0) == 404

    async def api_page_iter(self, url, page_size=100, _json=True, next_key=None, iter_key=None, **requests_kwargs):
        """
        An asynchronous generator function for iterating through paginated API data.

        This function continuously makes requests to a specified API URL, incrementing the page number
        or applying a custom pagination function, and yields the received data one page at a time.
        It is well-suited for APIs that provide paginated results.

        Args:
            url (str): The initial API URL. Can contain placeholders for 'page', 'page_size', and 'offset'.
            page_size (int, optional): The number of items per page. Defaults to 100.
            json (bool, optional): If True, attempts to deserialize the response content to a JSON object. Defaults to True.
            next_key (callable, optional): A function that takes the last page's data and returns the URL for the next page. Defaults to None.
            iter_key (callable, optional): A function that builds each new request based on the page number, page size, and offset. Defaults to a simple implementation that autoreplaces {page} and {page_size} in the url.
            **requests_kwargs: Arbitrary keyword arguments that will be forwarded to the HTTP request function.

        Yields:
            dict or httpx.Response: If 'json' is True, yields a dictionary containing the parsed JSON data. Otherwise, yields the raw HTTP response.

        Note:
            The loop will continue indefinitely unless manually stopped. Make sure to break out of the loop once the last page has been received.

        Examples:
            >>> agen = api_page_iter('https://api.example.com/data?page={page}&page_size={page_size}')
            >>> try:
            >>>     async for page in agen:
            >>>         subdomains = page["subdomains"]
            >>>         self.hugesuccess(subdomains)
            >>>         if not subdomains:
            >>>             break
            >>> finally:
            >>>     await agen.aclose()
        """
        page = 1
        offset = 0
        result = None
        if iter_key is None:
            iter_key = self._prepare_api_iter_req
        while 1:
            if result and callable(next_key):
                try:
                    new_url = next_key(result)
                except Exception as e:
                    self.debug(f"Failed to extract next page of results from {url}: {e}")
                    self.debug(traceback.format_exc())
            else:
                new_url, new_kwargs = iter_key(url, page, page_size, offset, **requests_kwargs)
            result = await self.api_request(new_url, **new_kwargs)
            if result is None:
                self.verbose(f"api_page_iter() got no response for {new_url}")
                break
            try:
                if _json:
                    result = result.json()
                yield result
            except Exception:
                self.warning(f'Error in api_page_iter() for url: "{new_url}"')
                self.trace(traceback.format_exc())