sources.helpers.rest_client.client
PageData Objects
class PageData(List[_T])
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()
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]]
Iterates over paginated API responses, yielding pages of data.
Arguments:
path
str - Endpoint path for the request, relative tobase_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 asstream
,verify
,proxies
,cert
,timeout
, andallow_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
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
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.