Skip to content

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 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
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 = 180  # seconds

    def __init__(
        self,
        base_url: str,
        auth: AuthBase | None = None,
        default_headers: dict[str, str] | None = None,
    ):
        """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.
        """
        if not base_url.endswith("/"):
            base_url += "/"
        self.base_url = base_url
        self.session = requests.Session()
        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, str] | None = None,
        data: dict[str, str] | None = None,
        json: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
        max_retries: int = 0,
    ) -> 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.

        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:
                    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(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)

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, str] | None

Default headers to include in requests.

 None
Source code in src/cloe_nessy/clients/api_client/api_client.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    base_url: str,
    auth: AuthBase | None = None,
    default_headers: dict[str, str] | None = None,
):
    """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.
    """
    if not base_url.endswith("/"):
        base_url += "/"
    self.base_url = base_url
    self.session = requests.Session()
    if default_headers:
        self.session.headers.update(default_headers)
    self.session.auth = auth

_make_request(method, endpoint, timeout=30, params=None, data=None, json=None, headers=None, max_retries=0)

Makes a request to the API endpoint.

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
timeout int

The timeout for the request in seconds.

 30
params dict[str, str] | None

The query parameters for the request.

 None
data dict[str, str] | None

The form data to include in the request.

 None
json dict[str, str] | None

The JSON data to include in the request.

 None
headers dict[str, str] | None

The headers to include in the request.

 None
max_retries int

The maximum number of retries for the request.

 0

Returns:

Name Type Description
APIResponse APIResponse

The response from the API.

Raises:

Type Description
APIClientError

If the request fails.
Source code in src/cloe_nessy/clients/api_client/api_client.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def _make_request(
    self,
    method: str,
    endpoint: str,
    timeout: int = 30,
    params: dict[str, str] | None = None,
    data: dict[str, str] | None = None,
    json: dict[str, str] | None = None,
    headers: dict[str, str] | None = None,
    max_retries: int = 0,
) -> 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.

    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:
                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(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")

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 
153
154
155
156
157
158
159
160
161
162
163
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 
117
118
119
120
121
122
123
124
125
126
127
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 
165
166
167
168
169
170
171
172
173
174
175
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 
129
130
131
132
133
134
135
136
137
138
139
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 
141
142
143
144
145
146
147
148
149
150
151
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 
177
178
179
180
181
182
183
184
185
186
187
188
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)