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 
 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
189
190
191
192
193
194
195
196
197
198
199
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 
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
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 
164
165
166
167
168
169
170
171
172
173
174
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 
128
129
130
131
132
133
134
135
136
137
138
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 
176
177
178
179
180
181
182
183
184
185
186
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 
140
141
142
143
144
145
146
147
148
149
150
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 
152
153
154
155
156
157
158
159
160
161
162
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 
188
189
190
191
192
193
194
195
196
197
198
199
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)