Index

APIClient

A standardized client for the interaction with APIs.

This class handles the communication with an API, including retries for specific status codes.

Attributes:

Name	Type	Description
RETRY_CODES	list[int]	List of HTTP status codes that should trigger a retry.
MAX_SLEEP_TIME	int	Maximum time to wait between retries, in seconds.
base_url		The base URL for the API.
session		The session object for making requests.

Source code in src/cloe_nessy/clients/api_client/api_client.py

class APIClient:
    """A standardized client for the interaction with APIs.

    This class handles the communication with an API, including retries for specific status codes.

    Attributes:
        RETRY_CODES: List of HTTP status codes that should trigger a retry.
        MAX_SLEEP_TIME: Maximum time to wait between retries, in seconds.
        base_url: The base URL for the API.
        session: The session object for making requests.
    """

    RETRY_CODES: list[int] = [
        HTTPStatus.TOO_MANY_REQUESTS,
        HTTPStatus.SERVICE_UNAVAILABLE,
        HTTPStatus.GATEWAY_TIMEOUT,
    ]

    MAX_SLEEP_TIME: int = 1800  # seconds

    def __init__(
        self,
        base_url: str,
        auth: AuthBase | None = None,
        default_headers: dict[str, Any] | None = None,
        pool_maxsize: int = 10,
    ):
        """Initializes the APIClient object.

        Args:
            base_url: The base URL for the API.
            auth: The authentication method for the API.
            default_headers: Default headers to include in requests.
            pool_maxsize: The maximum pool size for the HTTPAdapter (maximum number of connections to save in the pool).
        """
        if not base_url.endswith("/"):
            base_url += "/"
        self.base_url = base_url
        self.session = requests.Session()
        self.pool_maxsize = pool_maxsize
        adapter = HTTPAdapter(pool_maxsize=pool_maxsize)
        self.session.mount("https://", adapter)
        if default_headers:
            self.session.headers.update(default_headers)
        self.session.auth = auth

    def _make_request(
        self,
        method: str,
        endpoint: str,
        timeout: int = 30,
        params: dict[str, Any] | None = None,
        data: dict[str, Any] | None = None,
        json: dict[str, Any] | None = None,
        headers: dict[str, Any] | None = None,
        max_retries: int = 0,
        backoff_factor: int = 1,
        raise_for_status: bool = True,
    ) -> APIResponse:
        """Makes a request to the API endpoint.

        Args:
            method: The HTTP method to use for the request.
            endpoint: The endpoint to send the request to.
            timeout: The timeout for the request in seconds.
            params: The query parameters for the request.
            data: The form data to include in the request.
            json: The JSON data to include in the request.
            headers: The headers to include in the request.
            max_retries: The maximum number of retries for the request.
            backoff_factor: Factor for exponential backoff between retries.
            raise_for_status: Raise HTTPError, if one occurred.

        Returns:
            APIResponse: The response from the API.

        Raises:
            APIClientError: If the request fails.
        """
        url = urljoin(self.base_url, endpoint.strip("/"))
        params = params or {}
        data = data or {}
        json = json or {}
        headers = headers or {}

        for attempt in range(max_retries + 1):
            try:
                response = self.session.request(
                    method=method,
                    url=url,
                    timeout=timeout,
                    params=params,
                    data=data,
                    json=json,
                    headers=headers,
                )
                if response.status_code not in APIClient.RETRY_CODES:
                    if raise_for_status:
                        response.raise_for_status()
                    return APIResponse(response)
            except requests.exceptions.HTTPError as err:
                raise APIClientHTTPError(f"HTTP error occurred: {err}") from err
            except requests.exceptions.ConnectionError as err:
                if attempt < max_retries:
                    sleep_time = min(backoff_factor * (2**attempt), APIClient.MAX_SLEEP_TIME)
                    sleep(sleep_time)
                    continue
                raise APIClientConnectionError(f"Connection error occurred: {err}") from err
            except requests.exceptions.Timeout as err:
                raise APIClientTimeoutError(f"Timeout error occurred: {err}") from err
            except requests.exceptions.RequestException as err:
                raise APIClientError(f"An error occurred: {err}") from err
        raise APIClientError(f"The maximum configured retries of [ '{max_retries}' ] have been exceeded")

    def get(self, endpoint: str, **kwargs: Any) -> APIResponse:
        """Sends a GET request to the specified endpoint.

        Args:
            endpoint: The endpoint to send the request to.
            **kwargs: Additional arguments to pass to the request.

        Returns:
            APIResponse: The response from the API.
        """
        return self._make_request(method="GET", endpoint=endpoint, **kwargs)

    def post(self, endpoint: str, **kwargs: Any) -> APIResponse:
        """Sends a POST request to the specified endpoint.

        Args:
            endpoint: The endpoint to send the request to.
            **kwargs: Additional arguments to pass to the request.

        Returns:
            APIResponse: The response from the API.
        """
        return self._make_request(method="POST", endpoint=endpoint, **kwargs)

    def put(self, endpoint: str, **kwargs: Any) -> APIResponse:
        """Sends a PUT request to the specified endpoint.

        Args:
            endpoint: The endpoint to send the request to.
            **kwargs: Additional arguments to pass to the request.

        Returns:
            APIResponse: The response from the API.
        """
        return self._make_request(method="PUT", endpoint=endpoint, **kwargs)

    def delete(self, endpoint: str, **kwargs: Any) -> APIResponse:
        """Sends a DELETE request to the specified endpoint.

        Args:
            endpoint: The endpoint to send the request to.
            **kwargs: Additional arguments to pass to the request.

        Returns:
            APIResponse: The response from the API.
        """
        return self._make_request(method="DELETE", endpoint=endpoint, **kwargs)

    def patch(self, endpoint: str, **kwargs: Any) -> APIResponse:
        """Sends a PATCH request to the specified endpoint.

        Args:
            endpoint: The endpoint to send the request to.
            **kwargs: Additional arguments to pass to the request.

        Returns:
            APIResponse: The response from the API.
        """
        return self._make_request(method="PATCH", endpoint=endpoint, **kwargs)

    def request(self, method: str, endpoint: str, **kwargs: Any) -> APIResponse:
        """Sends a request to the specified endpoint with the specified method.

        Args:
            method: The HTTP method to use for the request.
            endpoint: The endpoint to send the request to.
            **kwargs: Additional arguments to pass to the request.

        Returns:
            APIResponse: The response from the API.
        """
        return self._make_request(method=method, endpoint=endpoint, **kwargs)

__init__(base_url, auth=None, default_headers=None, pool_maxsize=10)

Initializes the APIClient object.

Parameters:

Name	Type	Description	Default
base_url	str	The base URL for the API.	required
auth	AuthBase | None	The authentication method for the API.	None
default_headers	dict[str, Any] | None	Default headers to include in requests.	None
pool_maxsize	int	The maximum pool size for the HTTPAdapter (maximum number of connections to save in the pool).	10

Source code in src/cloe_nessy/clients/api_client/api_client.py

def __init__(
    self,
    base_url: str,
    auth: AuthBase | None = None,
    default_headers: dict[str, Any] | None = None,
    pool_maxsize: int = 10,
):
    """Initializes the APIClient object.

    Args:
        base_url: The base URL for the API.
        auth: The authentication method for the API.
        default_headers: Default headers to include in requests.
        pool_maxsize: The maximum pool size for the HTTPAdapter (maximum number of connections to save in the pool).
    """
    if not base_url.endswith("/"):
        base_url += "/"
    self.base_url = base_url
    self.session = requests.Session()
    self.pool_maxsize = pool_maxsize
    adapter = HTTPAdapter(pool_maxsize=pool_maxsize)
    self.session.mount("https://", adapter)
    if default_headers:
        self.session.headers.update(default_headers)
    self.session.auth = auth

delete(endpoint, **kwargs)

Sends a DELETE request to the specified endpoint.

Parameters:

Name	Type	Description	Default
endpoint	str	The endpoint to send the request to.	required
**kwargs	Any	Additional arguments to pass to the request.	{}

Returns:

Name	Type	Description
APIResponse	APIResponse	The response from the API.

Source code in src/cloe_nessy/clients/api_client/api_client.py

def delete(self, endpoint: str, **kwargs: Any) -> APIResponse:
    """Sends a DELETE request to the specified endpoint.

    Args:
        endpoint: The endpoint to send the request to.
        **kwargs: Additional arguments to pass to the request.

    Returns:
        APIResponse: The response from the API.
    """
    return self._make_request(method="DELETE", endpoint=endpoint, **kwargs)

get(endpoint, **kwargs)

Sends a GET request to the specified endpoint.

Parameters:

Name	Type	Description	Default
endpoint	str	The endpoint to send the request to.	required
**kwargs	Any	Additional arguments to pass to the request.	{}

Returns:

Name	Type	Description
APIResponse	APIResponse	The response from the API.

Source code in src/cloe_nessy/clients/api_client/api_client.py

def get(self, endpoint: str, **kwargs: Any) -> APIResponse:
    """Sends a GET request to the specified endpoint.

    Args:
        endpoint: The endpoint to send the request to.
        **kwargs: Additional arguments to pass to the request.

    Returns:
        APIResponse: The response from the API.
    """
    return self._make_request(method="GET", endpoint=endpoint, **kwargs)

patch(endpoint, **kwargs)

Sends a PATCH request to the specified endpoint.

Parameters:

Name	Type	Description	Default
endpoint	str	The endpoint to send the request to.	required
**kwargs	Any	Additional arguments to pass to the request.	{}

Returns:

Name	Type	Description
APIResponse	APIResponse	The response from the API.

Source code in src/cloe_nessy/clients/api_client/api_client.py

def patch(self, endpoint: str, **kwargs: Any) -> APIResponse:
    """Sends a PATCH request to the specified endpoint.

    Args:
        endpoint: The endpoint to send the request to.
        **kwargs: Additional arguments to pass to the request.

    Returns:
        APIResponse: The response from the API.
    """
    return self._make_request(method="PATCH", endpoint=endpoint, **kwargs)

post(endpoint, **kwargs)

Sends a POST request to the specified endpoint.

Parameters:

Name	Type	Description	Default
endpoint	str	The endpoint to send the request to.	required
**kwargs	Any	Additional arguments to pass to the request.	{}

Returns:

Name	Type	Description
APIResponse	APIResponse	The response from the API.

Source code in src/cloe_nessy/clients/api_client/api_client.py

def post(self, endpoint: str, **kwargs: Any) -> APIResponse:
    """Sends a POST request to the specified endpoint.

    Args:
        endpoint: The endpoint to send the request to.
        **kwargs: Additional arguments to pass to the request.

    Returns:
        APIResponse: The response from the API.
    """
    return self._make_request(method="POST", endpoint=endpoint, **kwargs)

put(endpoint, **kwargs)

Sends a PUT request to the specified endpoint.

Parameters:

Name	Type	Description	Default
endpoint	str	The endpoint to send the request to.	required
**kwargs	Any	Additional arguments to pass to the request.	{}

Returns:

Name	Type	Description
APIResponse	APIResponse	The response from the API.

Source code in src/cloe_nessy/clients/api_client/api_client.py

def put(self, endpoint: str, **kwargs: Any) -> APIResponse:
    """Sends a PUT request to the specified endpoint.

    Args:
        endpoint: The endpoint to send the request to.
        **kwargs: Additional arguments to pass to the request.

    Returns:
        APIResponse: The response from the API.
    """
    return self._make_request(method="PUT", endpoint=endpoint, **kwargs)

request(method, endpoint, **kwargs)

Sends a request to the specified endpoint with the specified method.

Parameters:

Name	Type	Description	Default
method	str	The HTTP method to use for the request.	required
endpoint	str	The endpoint to send the request to.	required
**kwargs	Any	Additional arguments to pass to the request.	{}

Returns:

Name	Type	Description
APIResponse	APIResponse	The response from the API.

Source code in src/cloe_nessy/clients/api_client/api_client.py

def request(self, method: str, endpoint: str, **kwargs: Any) -> APIResponse:
    """Sends a request to the specified endpoint with the specified method.

    Args:
        method: The HTTP method to use for the request.
        endpoint: The endpoint to send the request to.
        **kwargs: Additional arguments to pass to the request.

    Returns:
        APIResponse: The response from the API.
    """
    return self._make_request(method=method, endpoint=endpoint, **kwargs)

APIResponse

An abstracted response to implement parsing.

This class provides methods to parse the response from an API request.

Attributes:

Name	Type	Description
response		The original response object.
headers		The headers of the response.
status_code		The status code of the response.
content_type		The content type of the response.

Source code in src/cloe_nessy/clients/api_client/api_response.py

class APIResponse:
    """An abstracted response to implement parsing.

    This class provides methods to parse the response from an API request.

    Attributes:
        response: The original response object.
        headers: The headers of the response.
        status_code: The status code of the response.
        content_type: The content type of the response.
    """

    def __init__(self, response: requests.Response):
        """Initializes the APIResponse object.

        Args:
            response: The response object from an API request.
        """
        self.response = response
        self.headers = self.response.headers
        self.status_code = self.response.status_code
        self.url = self.response.url
        self.reason = self.response.reason
        self.elapsed = self.response.elapsed
        self.content_type = self.headers.get("Content-Type", "").lower()

    def to_dict(self, key: str | None = None) -> dict[str, Any]:
        """Parses the values from the response into a dictionary.

        Args:
            key: The key to return from the dictionary. If specified, the method
                will return the value associated with this key from the parsed dictionary.

        Returns:
            The response parsed to a dictionary. If a key is specified,
                the method returns the value associated with this key.

        Raises:
            KeyError: If the specified key is not found in the response.
            ValueError: If there is an error parsing the JSON response.
            Exception: For any other unexpected errors.
        """
        dict_response = {}
        try:
            if "application/json" in self.content_type:
                dict_response = self.response.json()
            else:
                # Handling of other response types can be added below.
                dict_response = {"value": self.response.text}

            if key:
                dict_response = {"value": dict_response[key]}
        except KeyError as err:
            raise KeyError(
                f"The key '{err.args[0]}' was not found in the response. Status code: {self.status_code}, "
                f"Headers: {self.headers}, Response: {dict_response}"
            ) from err
        except ValueError as err:
            raise ValueError(
                f"Error parsing JSON response: {err}. Status code: {self.status_code}, Headers: {self.headers}, "
                f"Response content: {self.response.text}"
            ) from err
        except Exception as err:
            raise APIClientError(
                f"An unexpected error occurred: {err}. Status code: {self.status_code}, Headers: {self.headers}, "
                f"Response content: {self.response.text}"
            ) from err
        return dict_response

__init__(response)

Initializes the APIResponse object.

Parameters:

Name	Type	Description	Default
response	Response	The response object from an API request.	required

Source code in src/cloe_nessy/clients/api_client/api_response.py

def __init__(self, response: requests.Response):
    """Initializes the APIResponse object.

    Args:
        response: The response object from an API request.
    """
    self.response = response
    self.headers = self.response.headers
    self.status_code = self.response.status_code
    self.url = self.response.url
    self.reason = self.response.reason
    self.elapsed = self.response.elapsed
    self.content_type = self.headers.get("Content-Type", "").lower()

to_dict(key=None)

Parses the values from the response into a dictionary.

Parameters:

Name	Type	Description	Default
key	str | None	The key to return from the dictionary. If specified, the method will return the value associated with this key from the parsed dictionary.	None

Returns:

Type	Description
dict[str, Any]	The response parsed to a dictionary. If a key is specified, the method returns the value associated with this key.

Raises:

Type	Description
KeyError	If the specified key is not found in the response.
ValueError	If there is an error parsing the JSON response.
Exception	For any other unexpected errors.

Source code in src/cloe_nessy/clients/api_client/api_response.py

def to_dict(self, key: str | None = None) -> dict[str, Any]:
    """Parses the values from the response into a dictionary.

    Args:
        key: The key to return from the dictionary. If specified, the method
            will return the value associated with this key from the parsed dictionary.

    Returns:
        The response parsed to a dictionary. If a key is specified,
            the method returns the value associated with this key.

    Raises:
        KeyError: If the specified key is not found in the response.
        ValueError: If there is an error parsing the JSON response.
        Exception: For any other unexpected errors.
    """
    dict_response = {}
    try:
        if "application/json" in self.content_type:
            dict_response = self.response.json()
        else:
            # Handling of other response types can be added below.
            dict_response = {"value": self.response.text}

        if key:
            dict_response = {"value": dict_response[key]}
    except KeyError as err:
        raise KeyError(
            f"The key '{err.args[0]}' was not found in the response. Status code: {self.status_code}, "
            f"Headers: {self.headers}, Response: {dict_response}"
        ) from err
    except ValueError as err:
        raise ValueError(
            f"Error parsing JSON response: {err}. Status code: {self.status_code}, Headers: {self.headers}, "
            f"Response content: {self.response.text}"
        ) from err
    except Exception as err:
        raise APIClientError(
            f"An unexpected error occurred: {err}. Status code: {self.status_code}, Headers: {self.headers}, "
            f"Response content: {self.response.text}"
        ) from err
    return dict_response

PaginationConfig

Bases: BaseModel

Configuration model for pagination options.

Source code in src/cloe_nessy/clients/api_client/pagination_config.py

class PaginationConfig(BaseModel):
    """Configuration model for pagination options."""

    strategy: str = Field(..., description="Pagination strategy (limit_offset, page_based, cursor_based, etc.)")
    check_field: str | None = Field(None, description="Field to check for emptiness of response.")
    next_page_field: str | None = Field(None, description="Field that indicates there is a next page.")
    limit_field: str | None = Field(
        None, description="Name of the limit parameter field for items per page or request."
    )
    offset_field: str | None = Field(
        None, description="Name of the offset parameter field for items per page or request."
    )
    page_field: str | None = Field(None, description="Name of the page parameter field.")
    max_page: int = Field(-1, description="Amount of pages to fetch. If not set, will fetch all available data.")
    pages_per_array_limit: int = Field(-1, description="Maximum number of pages per array.")
    preliminary_probe: bool = Field(
        False, description="Whether to perform a preliminary probe to determine the total number of pages."
    )

    @field_validator("strategy", mode="before")
    @classmethod
    def _validate_strategy(cls, v: str) -> str:
        """Validates the pagination strategy."""
        supported_strategies = ["limit_offset", "page_based", "cursor_based", "time_based"]
        if v not in supported_strategies:
            if v in ["cursor_based", "time_based"]:
                raise NotImplementedError("cursor_based and time_based are not yet supported.")
            supported_str = ", ".join(supported_strategies)
            raise ValueError(f"Unsupported pagination strategy: {v}. Supported strategies: {supported_str}")
        return v

    @model_validator(mode="after")
    def _validate_strategy_config(self) -> Self:
        """Validates the configuration of the pagination strategy."""
        if self.strategy == "limit_offset" and any(field is None for field in [self.limit_field, self.offset_field]):
            raise ValueError(f"Both <limit_field> and <offset_field> must be set for strategy '{self.strategy}'")
        if self.strategy == "page_based" and self.page_field is None:
            raise ValueError(f"<page_field> must be set for strategy '{self.strategy}'")
        return self

PaginationConfigData

Bases: TypedDict

Top-level config (what your Pydantic model or dict can accept).

Source code in src/cloe_nessy/clients/api_client/pagination_config.py

class PaginationConfigData(TypedDict, total=False):
    """Top-level config (what your Pydantic model or dict can accept)."""

    strategy: str  # "limit_offset" | "page_based" | ...
    # strategy-specific fields:
    limit_field: str
    offset_field: str
    page_field: str
    # shared/advanced fields:
    check_field: str | None
    next_page_field: str | None
    max_page: int
    pages_per_array_limit: int
    preliminary_probe: bool