|
| 1 | +import asyncio |
1 | 2 | import logging |
| 3 | +import random |
2 | 4 | from contextlib import AbstractAsyncContextManager |
| 5 | +from datetime import UTC, datetime |
| 6 | +from email.utils import parsedate_to_datetime |
3 | 7 | from types import TracebackType |
4 | 8 | from typing import Any, cast |
5 | 9 |
|
6 | 10 | import httpx |
7 | 11 | from pydantic import SecretStr |
8 | 12 |
|
| 13 | +from .config import RetryConfig |
9 | 14 | from .exceptions import NetworkError, TimeoutError, raise_for_status |
10 | 15 |
|
11 | 16 | log = logging.getLogger(__name__) |
12 | 17 |
|
13 | 18 |
|
| 19 | +def _parse_retry_after(value: str | None) -> float | None: |
| 20 | + """Parse a Retry-After header: delta-seconds or an HTTP-date.""" |
| 21 | + if not value: |
| 22 | + return None |
| 23 | + if value.isdigit(): |
| 24 | + return float(value) |
| 25 | + try: |
| 26 | + retry_at = parsedate_to_datetime(value) |
| 27 | + except (TypeError, ValueError): |
| 28 | + return None |
| 29 | + return max(0.0, (retry_at - datetime.now(UTC)).total_seconds()) |
| 30 | + |
| 31 | + |
14 | 32 | class APIHttpClient: |
15 | 33 | """Thin async transport around httpx. |
16 | 34 |
|
17 | 35 | Receives its full configuration from the caller (the SDK resolves |
18 | 36 | arguments, environment variables, and defaults) and owns nothing but |
19 | | - the connection lifecycle and error mapping. |
| 37 | + the connection lifecycle, error mapping, and opt-in retries. |
20 | 38 | """ |
21 | 39 |
|
22 | | - def __init__(self, *, token: SecretStr, base_url: str, timeout: httpx.Timeout): |
| 40 | + def __init__( |
| 41 | + self, |
| 42 | + *, |
| 43 | + token: SecretStr, |
| 44 | + base_url: str, |
| 45 | + timeout: httpx.Timeout, |
| 46 | + retry: RetryConfig | None = None, |
| 47 | + ): |
23 | 48 | self._token = token.get_secret_value() |
24 | 49 | self._base_url = base_url |
25 | 50 | self._client: httpx.AsyncClient | None = None |
| 51 | + self._retry = retry if retry is not None else RetryConfig() |
26 | 52 |
|
27 | 53 | self._timeout_config = timeout |
28 | 54 | self._client_config: dict[str, Any] = { |
@@ -92,28 +118,66 @@ async def request( |
92 | 118 | log.debug(f"Request: {method} {endpoint}") |
93 | 119 | log.debug(f"Request kwargs: {kwargs}") |
94 | 120 |
|
95 | | - try: |
96 | | - response = await client.request(method, endpoint, **kwargs) |
| 121 | + retryable_method = method.upper() in self._retry.retry_methods |
| 122 | + attempts = self._retry.max_retries + 1 if retryable_method else 1 |
| 123 | + |
| 124 | + for attempt in range(attempts): |
| 125 | + last_attempt = attempt == attempts - 1 |
| 126 | + try: |
| 127 | + response = await client.request(method, endpoint, **kwargs) |
| 128 | + except httpx.TimeoutException as e: |
| 129 | + if not last_attempt: |
| 130 | + await self._sleep_before_retry(attempt, None, method, endpoint) |
| 131 | + continue |
| 132 | + log.error(f"Request timeout for {method} {endpoint}: {e}") |
| 133 | + raise TimeoutError(f"Request to {endpoint} timed out.") from e |
| 134 | + except httpx.ConnectError as e: |
| 135 | + if not last_attempt: |
| 136 | + await self._sleep_before_retry(attempt, None, method, endpoint) |
| 137 | + continue |
| 138 | + log.error(f"Connection error for {method} {endpoint}: {e}") |
| 139 | + raise NetworkError( |
| 140 | + f"Failed to connect to the API: {e}", |
| 141 | + original_error=e, |
| 142 | + ) from e |
| 143 | + except httpx.RequestError as e: |
| 144 | + log.error(f"Network error for {method} {endpoint}: {e}") |
| 145 | + raise NetworkError( |
| 146 | + f"A network error occurred: {e}", |
| 147 | + original_error=e, |
| 148 | + ) from e |
| 149 | + |
97 | 150 | log.debug( |
98 | 151 | f"Response: {response.status_code} {response.reason_phrase} " |
99 | 152 | f"for {method} {endpoint}" |
100 | 153 | ) |
101 | 154 |
|
| 155 | + if response.status_code in self._retry.retry_statuses and not last_attempt: |
| 156 | + await self._sleep_before_retry(attempt, response, method, endpoint) |
| 157 | + continue |
| 158 | + |
102 | 159 | raise_for_status(response) |
103 | 160 | return response |
104 | 161 |
|
105 | | - except httpx.TimeoutException as e: |
106 | | - log.error(f"Request timeout for {method} {endpoint}: {e}") |
107 | | - raise TimeoutError(f"Request to {endpoint} timed out.") from e |
108 | | - except httpx.ConnectError as e: |
109 | | - log.error(f"Connection error for {method} {endpoint}: {e}") |
110 | | - raise NetworkError( |
111 | | - f"Failed to connect to the API: {e}", |
112 | | - original_error=e, |
113 | | - ) from e |
114 | | - except httpx.RequestError as e: |
115 | | - log.error(f"Network error for {method} {endpoint}: {e}") |
116 | | - raise NetworkError( |
117 | | - f"A network error occurred: {e}", |
118 | | - original_error=e, |
119 | | - ) from e |
| 162 | + raise AssertionError("unreachable: retry loop must return or raise") |
| 163 | + |
| 164 | + async def _sleep_before_retry( |
| 165 | + self, |
| 166 | + attempt: int, |
| 167 | + response: httpx.Response | None, |
| 168 | + method: str, |
| 169 | + endpoint: str, |
| 170 | + ) -> None: |
| 171 | + delay = None |
| 172 | + if response is not None: |
| 173 | + delay = _parse_retry_after(response.headers.get("Retry-After")) |
| 174 | + if delay is None: |
| 175 | + delay = self._retry.backoff_factor * (2**attempt) |
| 176 | + # Jitter to avoid thundering herds; not security-relevant. |
| 177 | + delay += random.uniform(0, self._retry.backoff_factor / 2) # nosec B311 |
| 178 | + |
| 179 | + log.debug( |
| 180 | + f"Retrying {method} {endpoint} in {delay:.2f}s " |
| 181 | + f"(attempt {attempt + 1}/{self._retry.max_retries})" |
| 182 | + ) |
| 183 | + await asyncio.sleep(delay) |
0 commit comments