Coverage for src/taipanstack/bridges/http_bridge.py: 100%
157 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-10-13 20:52 +0000
1"""
2HTTP Bridge — safe httpx client with SSRF protection and resilience.
4Wraps ``httpx.AsyncClient`` with TaipanStack's ``guard_ssrf``,
5retry, and circuit breaker integrations. All outbound URLs are
6validated against SSRF before the request is sent.
7"""
9from __future__ import annotations
11import asyncio
12import logging
13from collections.abc import Awaitable, Callable, Mapping, Sequence
14from typing import TYPE_CHECKING, TypeAlias, cast
16from typing_extensions import TypedDict, Unpack
18if TYPE_CHECKING:
19 import httpx
22from taipanstack.core.result import Err, Ok, Result
23from taipanstack.resilience.circuit_breaker import (
24 CircuitBreaker,
25 CircuitBreakerError,
26)
27from taipanstack.resilience.retry import RetryConfig, calculate_delay
28from taipanstack.security.guards import guard_ssrf
30logger = logging.getLogger("taipanstack.bridges.http")
33# JsonType definition for robust typing
34JsonType: TypeAlias = dict[str, object] | list[object] | str | int | float | bool | None
37class HttpRequestKwargs(TypedDict, total=False):
38 """Type definitions for HTTP request kwargs."""
40 content: bytes | str | None
41 data: (
42 dict[str, str | int | float | bool | None]
43 | list[tuple[str, str]]
44 | bytes
45 | str
46 | None
47 )
48 files: dict[str, bytes | tuple[str, bytes]]
49 json: JsonType
50 params: (
51 dict[
52 str,
53 str | int | float | bool | None | Sequence[str | int | float | bool | None],
54 ]
55 | list[tuple[str, str | int | float | bool | None]]
56 | str
57 | bytes
58 | None
59 )
60 headers: dict[str, str]
61 cookies: dict[str, str]
62 auth: tuple[str, str]
63 follow_redirects: bool
64 extensions: dict[str, object]
67class HttpClientKwargs(TypedDict, total=False):
68 """Type definitions for HTTP client kwargs."""
70 base_url: str
71 headers: dict[str, str]
72 cookies: dict[str, str]
73 verify: bool | str
74 cert: str | tuple[str, str] | tuple[str, str, str]
75 http1: bool
76 http2: bool
77 proxy: str
78 mounts: Mapping[str, httpx.AsyncBaseTransport | None]
79 follow_redirects: bool
80 max_redirects: int
81 event_hooks: dict[str, list[Callable[..., Awaitable[object]]]]
82 trust_env: bool
83 default_encoding: str | Callable[[bytes], str]
86# --- optional httpx import ------------------------------------------------
88try:
89 import httpx
91 _HAS_HTTPX = True
92except ImportError:
93 _HAS_HTTPX = False
95if TYPE_CHECKING:
96 import httpx
98# Default status codes that trigger a retry
99_RETRYABLE_STATUS_CODES: frozenset[int] = frozenset({500, 502, 503, 504, 429})
102def _check_circuit_breaker(
103 circuit_breaker: CircuitBreaker,
104) -> CircuitBreakerError | None:
105 """Check circuit breaker state and return error if OPEN.
107 Args:
108 circuit_breaker: The circuit breaker to check.
110 Returns:
111 ``CircuitBreakerError`` if open, ``None`` otherwise.
113 """
114 if not circuit_breaker._should_attempt():
115 return CircuitBreakerError(
116 f"Circuit '{circuit_breaker.name}' is open",
117 state=circuit_breaker.state,
118 )
119 return None
122def _check_ssrf(url: str, ssrf_protection: bool) -> Result[None, Exception]:
123 """Validate URL against SSRF if protection is enabled.
125 Args:
126 url: The URL to validate.
127 ssrf_protection: Whether SSRF protection is enabled.
129 Returns:
130 ``Ok(None)`` if valid or protection disabled, ``Err(Exception)`` otherwise.
132 """
133 if not ssrf_protection:
134 return Ok(None)
136 ssrf_result = guard_ssrf(url)
137 if isinstance(ssrf_result, Err):
138 return Err(ssrf_result.err_value)
139 return Ok(None)
142def _should_retry_status(
143 response: httpx.Response,
144 retry_config: RetryConfig | None,
145 retryable_status_codes: frozenset[int],
146 attempt: int,
147 max_attempts: int,
148) -> bool:
149 """Determine if a request should be retried based on status code."""
150 return (
151 retry_config is not None
152 and response.status_code in retryable_status_codes
153 and attempt < max_attempts
154 )
157async def _handle_http_exception(
158 exc: Exception,
159 attempt: int,
160 max_attempts: int,
161 retry_config: RetryConfig | None,
162 circuit_breaker: CircuitBreaker | None,
163) -> bool:
164 """Handle exception and return True if we should retry, False otherwise."""
165 if circuit_breaker is not None:
166 circuit_breaker._record_failure(exc)
167 if retry_config is not None and attempt < max_attempts:
168 delay = calculate_delay(attempt, retry_config)
169 await asyncio.sleep(min(delay, 3600.0))
170 return True
171 return False
174async def _execute_single_attempt(
175 request_func: Callable[[], Awaitable[httpx.Response]],
176 retry_config: RetryConfig | None,
177 circuit_breaker: CircuitBreaker | None,
178 retryable_status_codes: frozenset[int],
179 attempt: int,
180 max_attempts: int,
181) -> Result[httpx.Response, Exception] | Exception | bool:
182 """Execute a single HTTP attempt.
184 Returns Result on success/final failure, Exception if it failed and we
185 should record it, or True if we should just retry because of status code.
186 """
187 try:
188 response = await request_func()
189 if _should_retry_status(
190 response,
191 retry_config,
192 retryable_status_codes,
193 attempt,
194 max_attempts,
195 ):
196 # Config is not None if we reach here
197 delay = calculate_delay(attempt, cast(RetryConfig, retry_config))
198 await asyncio.sleep(min(delay, 3600.0))
199 return True
200 return Ok(response)
201 except Exception as exc:
202 should_retry = await _handle_http_exception(
203 exc,
204 attempt,
205 max_attempts,
206 retry_config,
207 circuit_breaker,
208 )
209 if should_retry:
210 return exc
211 return Err(exc)
214def _process_attempt_outcome(
215 outcome: Result[httpx.Response, Exception] | Exception | bool,
216 last_error: Exception,
217) -> tuple[bool, Exception, Result[httpx.Response, Exception] | None]:
218 if isinstance(outcome, (Ok, Err)):
219 return True, last_error, outcome
220 if isinstance(outcome, Exception):
221 return False, outcome, None
222 return False, last_error, None
225def _get_max_attempts(retry_config: RetryConfig | None) -> int:
226 return retry_config.max_attempts if retry_config is not None else 1
229async def _execute_with_retries(
230 request_func: Callable[[], Awaitable[httpx.Response]],
231 retry_config: RetryConfig | None,
232 circuit_breaker: CircuitBreaker | None,
233 retryable_status_codes: frozenset[int],
234) -> Result[httpx.Response, Exception]:
235 """Execute a request function with optional retries and circuit breaker.
237 Args:
238 request_func: Async callable that performs the request.
239 retry_config: Optional retry configuration.
240 circuit_breaker: Optional circuit breaker.
241 retryable_status_codes: Status codes to retry on.
243 Returns:
244 ``Ok(Response)`` on success, ``Err`` on failure.
246 """
247 max_attempts = _get_max_attempts(retry_config)
248 last_error: Exception = RuntimeError("Request failed")
250 for attempt in range(1, max_attempts + 1):
251 outcome = await _execute_single_attempt(
252 request_func,
253 retry_config,
254 circuit_breaker,
255 retryable_status_codes,
256 attempt,
257 max_attempts,
258 )
259 is_finished, last_error, result = _process_attempt_outcome(outcome, last_error)
260 if is_finished and result is not None:
261 return result
263 return Err(last_error)
266def _check_preconditions(
267 url: str,
268 ssrf_protection: bool,
269 circuit_breaker: CircuitBreaker | None,
270) -> Err[Exception] | None:
271 """Check SSRF and circuit breaker preconditions."""
272 ssrf_check = _check_ssrf(url, ssrf_protection)
273 if isinstance(ssrf_check, Err):
274 return ssrf_check
276 if circuit_breaker is not None:
277 cb_err: Exception | None = _check_circuit_breaker(circuit_breaker)
278 if cb_err is not None:
279 return Err(cb_err)
281 return None
284async def safe_request(
285 method: str,
286 url: str,
287 *,
288 ssrf_protection: bool = True,
289 retry_config: RetryConfig | None = None,
290 circuit_breaker: CircuitBreaker | None = None,
291 retryable_status_codes: frozenset[int] = _RETRYABLE_STATUS_CODES,
292 timeout: float | None = 10.0,
293 **kwargs: Unpack[HttpRequestKwargs],
294) -> Result[httpx.Response, Exception]:
295 """Perform a one-shot HTTP request with safety features.
297 Args:
298 method: HTTP method (GET, POST, etc.).
299 url: Target URL.
300 ssrf_protection: Validate URL against SSRF.
301 retry_config: Optional retry configuration.
302 circuit_breaker: Optional circuit breaker.
303 retryable_status_codes: Status codes that trigger retries.
304 timeout: Explicit timeout in seconds (default: 10.0).
305 **kwargs: Passed to ``httpx.AsyncClient.request``.
307 Returns:
308 ``Ok(Response)`` on success, ``Err`` on failure.
310 """
311 if not _HAS_HTTPX:
312 return Err(
313 ImportError(
314 "httpx is required for HTTP bridge. "
315 "Install with: pip install taipanstack[bridges-http]",
316 ),
317 )
319 precondition_err = _check_preconditions(url, ssrf_protection, circuit_breaker)
320 if precondition_err is not None:
321 return precondition_err
323 async def _do_request() -> httpx.Response:
324 async with httpx.AsyncClient(timeout=timeout) as client: # nosemgrep
325 request_func = cast( # type: ignore[misc]
326 Callable[..., Awaitable[httpx.Response]],
327 client.request,
328 )
329 response = await request_func(method, url, **kwargs)
330 return response
332 return await _execute_with_retries(
333 _do_request,
334 retry_config,
335 circuit_breaker,
336 retryable_status_codes,
337 )
340class SafeHttpClient:
341 """Async context manager wrapping httpx with TaipanStack safety.
343 Args:
344 ssrf_protection: Enable SSRF validation on all requests.
345 retry_config: Retry configuration for transient failures.
346 circuit_breaker: Optional circuit breaker for all requests.
347 retryable_status_codes: HTTP status codes to retry on.
348 **client_kwargs: Passed to ``httpx.AsyncClient``.
350 Example:
351 >>> async with SafeHttpClient() as client:
352 ... result = await client.get("https://api.example.com/data")
353 ... if isinstance(result, Ok): print(result.unwrap().json())
354 ... else: print(f"Error: {result.unwrap_err()}")
356 """
358 def __init__(
359 self,
360 *,
361 ssrf_protection: bool = True,
362 retry_config: RetryConfig | None = None,
363 circuit_breaker: CircuitBreaker | None = None,
364 retryable_status_codes: frozenset[int] = _RETRYABLE_STATUS_CODES,
365 timeout: float = 10.0,
366 **client_kwargs: Unpack[HttpClientKwargs],
367 ) -> None:
368 """Initialize the safe HTTP client.
370 Args:
371 ssrf_protection: Enable SSRF validation.
372 retry_config: Retry configuration.
373 circuit_breaker: Circuit breaker instance.
374 retryable_status_codes: Status codes to retry.
375 **client_kwargs: Keyword args for httpx.AsyncClient.
376 Default timeout is 10.0 seconds if not provided.
378 """
379 self._ssrf_protection = ssrf_protection
380 self._retry_config = retry_config
381 self._circuit_breaker = circuit_breaker
382 self._retryable_status_codes = retryable_status_codes
383 self._client_kwargs = client_kwargs # type: ignore[misc]
384 self._timeout = timeout
385 self._client: httpx.AsyncClient | None = None
387 async def __aenter__(self) -> SafeHttpClient:
388 """Enter the async context manager."""
389 if not _HAS_HTTPX:
390 msg = (
391 "httpx is required for SafeHttpClient. "
392 "Install with: pip install taipanstack[bridges-http]"
393 )
394 raise ImportError(msg)
395 self._client = httpx.AsyncClient(
396 timeout=self._timeout,
397 **self._client_kwargs, # type: ignore[misc]
398 ) # nosemgrep
399 return self
401 async def __aexit__(
402 self,
403 _exc_type: type[BaseException] | None,
404 _exc_val: BaseException | None,
405 _exc_tb: object,
406 ) -> None:
407 """Exit the async context manager."""
408 if self._client is not None:
409 await self._client.aclose()
410 self._client = None
412 async def request(
413 self,
414 method: str,
415 url: str,
416 **kwargs: Unpack[HttpRequestKwargs],
417 ) -> Result[httpx.Response, Exception]:
418 """Send an HTTP request with safety features.
420 Args:
421 method: HTTP method.
422 url: Target URL.
423 **kwargs: Passed to the underlying client.
425 Returns:
426 ``Ok(Response)`` on success, ``Err`` on failure.
428 """
429 if self._client is None:
430 return Err(RuntimeError("Client not initialised. Use 'async with'."))
432 precondition_err = _check_preconditions(
433 url, self._ssrf_protection, self._circuit_breaker
434 )
435 if precondition_err is not None:
436 return precondition_err
438 async def _do_request() -> httpx.Response:
439 # We explicitly verified client is not None above
440 client: httpx.AsyncClient = cast(httpx.AsyncClient, self._client)
441 request_func = cast( # type: ignore[misc]
442 Callable[..., Awaitable[httpx.Response]],
443 client.request,
444 )
445 response = await request_func(method, url, **kwargs)
446 return response
448 return await _execute_with_retries(
449 _do_request,
450 self._retry_config,
451 self._circuit_breaker,
452 self._retryable_status_codes,
453 )
455 async def get(
456 self,
457 url: str,
458 **kw: Unpack[HttpRequestKwargs],
459 ) -> Result[httpx.Response, Exception]:
460 """Send a GET request."""
461 return await self.request("GET", url, **kw)
463 async def post(
464 self,
465 url: str,
466 **kw: Unpack[HttpRequestKwargs],
467 ) -> Result[httpx.Response, Exception]:
468 """Send a POST request."""
469 return await self.request("POST", url, **kw)
471 async def put(
472 self,
473 url: str,
474 **kw: Unpack[HttpRequestKwargs],
475 ) -> Result[httpx.Response, Exception]:
476 """Send a PUT request."""
477 return await self.request("PUT", url, **kw)
479 async def delete(
480 self,
481 url: str,
482 **kw: Unpack[HttpRequestKwargs],
483 ) -> Result[httpx.Response, Exception]:
484 """Send a DELETE request."""
485 return await self.request("DELETE", url, **kw)
487 async def patch(
488 self,
489 url: str,
490 **kw: Unpack[HttpRequestKwargs],
491 ) -> Result[httpx.Response, Exception]:
492 """Send a PATCH request."""
493 return await self.request("PATCH", url, **kw)