sources.helpers.rest_client.client

PageData Objects

class PageData(List[_T])

[view_source]

A list of elements in a single page of results with attached request context.

The context allows to inspect the response, paginator and authenticator, modify the request

RESTClient Objects

class RESTClient()

[view_source]

A generic REST client for making requests to an API with support for pagination and authentication.

Arguments:

• base_url str - The base URL of the API to make requests to.
• headers Optional[Dict[str, str]] - Default headers to include in all requests.
• auth Optional[AuthBase] - Authentication configuration for all requests.
• paginator Optional[BasePaginator] - Default paginator for handling paginated responses.
• data_selector Optional[jsonpath.TJsonPath] - JSONPath selector for extracting data from responses.
• session BaseSession - HTTP session for making requests.
• paginator_factory Optional[PaginatorFactory] - Factory for creating paginator instances, used for detecting paginators.

paginate

def paginate(path: str = "",
             method: HTTPMethodBasic = "GET",
             params: Optional[Dict[str, Any]] = None,
             json: Optional[Dict[str, Any]] = None,
             auth: Optional[AuthBase] = None,
             paginator: Optional[BasePaginator] = None,
             data_selector: Optional[jsonpath.TJsonPath] = None,
             hooks: Optional[Hooks] = None,
             **kwargs: Any) -> Iterator[PageData[Any]]

[view_source]

Iterates over paginated API responses, yielding pages of data.

Arguments:

• path str - Endpoint path for the request, relative to base_url.
• method HTTPMethodBasic - HTTP method for the request, defaults to 'get'.
• params Optional[Dict[str, Any]] - URL parameters for the request.
• json Optional[Dict[str, Any]] - JSON payload for the request.
• auth Optional[AuthBase - Authentication configuration for the request.
• paginator Optional[BasePaginator] - Paginator instance for handling pagination logic.
• data_selector Optional[jsonpath.TJsonPath] - JSONPath selector for extracting data from the response.
• hooks Optional[Hooks] - Hooks to modify request/response objects. Note that when hooks are not provided, the default behavior is to raise an exception on error status codes.
• **kwargs Any - Optional arguments to that the Request library accepts, such as stream, verify, proxies, cert, timeout, and allow_redirects.

Yields:

• PageData[Any] - A page of data from the paginated API response, along with request and response context.

Raises:

• HTTPError - If the response status code is not a success code. This is raised by default when hooks are not provided.

Example:

    client = RESTClient(base_url="https://api.example.com")
    for page in client.paginate("/search", method="post", json={"query": "foo"}):
        print(page)

detect_data_selector

def detect_data_selector(response: Response) -> str

[view_source]

Detects a path to page data in response. If there's no paging detected, returns "$" which will select full response

Returns:

• str - a json path to the page data.

detect_paginator

def detect_paginator(response: Response, data: Any) -> BasePaginator

[view_source]

Detects a paginator for the response and returns it.

Arguments:

• response Response - The response to detect the paginator for.
• data_selector data_selector - Path to paginated data or $ if paginated data not detected

Returns:

• BasePaginator - The paginator instance that was detected.