Coverage for src/taipanstack/bridges/http_bridge.py: 100%
150 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 15:01 +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)
214async def _execute_with_retries(
215 request_func: Callable[[], Awaitable[httpx.Response]],
216 retry_config: RetryConfig | None,
217 circuit_breaker: CircuitBreaker | None,
218 retryable_status_codes: frozenset[int],
219) -> Result[httpx.Response, Exception]:
220 """Execute a request function with optional retries and circuit breaker.
222 Args:
223 request_func: Async callable that performs the request.
224 retry_config: Optional retry configuration.
225 circuit_breaker: Optional circuit breaker.
226 retryable_status_codes: Status codes to retry on.
228 Returns:
229 ``Ok(Response)`` on success, ``Err`` on failure.
231 """
232 max_attempts = retry_config.max_attempts if retry_config is not None else 1
233 last_error: Exception = RuntimeError("Request failed")
235 for attempt in range(1, max_attempts + 1):
236 outcome = await _execute_single_attempt(
237 request_func,
238 retry_config,
239 circuit_breaker,
240 retryable_status_codes,
241 attempt,
242 max_attempts,
243 )
244 if isinstance(outcome, (Ok, Err)):
245 return outcome
246 if isinstance(outcome, Exception):
247 last_error = outcome
249 return Err(last_error)
252def _check_preconditions(
253 url: str,
254 ssrf_protection: bool,
255 circuit_breaker: CircuitBreaker | None,
256) -> Err[Exception] | None:
257 """Check SSRF and circuit breaker preconditions."""
258 ssrf_check = _check_ssrf(url, ssrf_protection)
259 if isinstance(ssrf_check, Err):
260 return ssrf_check
262 if circuit_breaker is not None:
263 cb_err: Exception | None = _check_circuit_breaker(circuit_breaker)
264 if cb_err is not None:
265 return Err(cb_err)
267 return None
270async def safe_request(
271 method: str,
272 url: str,
273 *,
274 ssrf_protection: bool = True,
275 retry_config: RetryConfig | None = None,
276 circuit_breaker: CircuitBreaker | None = None,
277 retryable_status_codes: frozenset[int] = _RETRYABLE_STATUS_CODES,
278 timeout: float | None = 10.0,
279 **kwargs: Unpack[HttpRequestKwargs],
280) -> Result[httpx.Response, Exception]:
281 """Perform a one-shot HTTP request with safety features.
283 Args:
284 method: HTTP method (GET, POST, etc.).
285 url: Target URL.
286 ssrf_protection: Validate URL against SSRF.
287 retry_config: Optional retry configuration.
288 circuit_breaker: Optional circuit breaker.
289 retryable_status_codes: Status codes that trigger retries.
290 timeout: Explicit timeout in seconds (default: 10.0).
291 **kwargs: Passed to ``httpx.AsyncClient.request``.
293 Returns:
294 ``Ok(Response)`` on success, ``Err`` on failure.
296 """
297 if not _HAS_HTTPX:
298 return Err(
299 ImportError(
300 "httpx is required for HTTP bridge. "
301 "Install with: pip install taipanstack[bridges-http]",
302 ),
303 )
305 precondition_err = _check_preconditions(url, ssrf_protection, circuit_breaker)
306 if precondition_err is not None:
307 return precondition_err
309 async def _do_request() -> httpx.Response:
310 async with httpx.AsyncClient(timeout=timeout) as client: # nosemgrep
311 request_func = cast(
312 Callable[..., Awaitable[httpx.Response]],
313 client.request,
314 )
315 response = await request_func(method, url, **kwargs)
316 return response
318 return await _execute_with_retries(
319 _do_request,
320 retry_config,
321 circuit_breaker,
322 retryable_status_codes,
323 )
326class SafeHttpClient:
327 """Async context manager wrapping httpx with TaipanStack safety.
329 Args:
330 ssrf_protection: Enable SSRF validation on all requests.
331 retry_config: Retry configuration for transient failures.
332 circuit_breaker: Optional circuit breaker for all requests.
333 retryable_status_codes: HTTP status codes to retry on.
334 **client_kwargs: Passed to ``httpx.AsyncClient``.
336 Example:
337 >>> async with SafeHttpClient() as client:
338 ... result = await client.get("https://api.example.com/data")
339 ... if isinstance(result, Ok): print(result.unwrap().json())
340 ... else: print(f"Error: {result.unwrap_err()}")
342 """
344 def __init__(
345 self,
346 *,
347 ssrf_protection: bool = True,
348 retry_config: RetryConfig | None = None,
349 circuit_breaker: CircuitBreaker | None = None,
350 retryable_status_codes: frozenset[int] = _RETRYABLE_STATUS_CODES,
351 timeout: float = 10.0,
352 **client_kwargs: Unpack[HttpClientKwargs],
353 ) -> None:
354 """Initialize the safe HTTP client.
356 Args:
357 ssrf_protection: Enable SSRF validation.
358 retry_config: Retry configuration.
359 circuit_breaker: Circuit breaker instance.
360 retryable_status_codes: Status codes to retry.
361 **client_kwargs: Keyword args for httpx.AsyncClient.
362 Default timeout is 10.0 seconds if not provided.
364 """
365 self._ssrf_protection = ssrf_protection
366 self._retry_config = retry_config
367 self._circuit_breaker = circuit_breaker
368 self._retryable_status_codes = retryable_status_codes
369 self._client_kwargs = client_kwargs
370 self._timeout = timeout
371 self._client: httpx.AsyncClient | None = None
373 async def __aenter__(self) -> SafeHttpClient:
374 """Enter the async context manager."""
375 if not _HAS_HTTPX:
376 msg = (
377 "httpx is required for SafeHttpClient. "
378 "Install with: pip install taipanstack[bridges-http]"
379 )
380 raise ImportError(msg)
381 self._client = httpx.AsyncClient(
382 timeout=self._timeout,
383 **self._client_kwargs,
384 ) # nosemgrep
385 return self
387 async def __aexit__(
388 self,
389 _exc_type: type[BaseException] | None,
390 _exc_val: BaseException | None,
391 _exc_tb: object,
392 ) -> None:
393 """Exit the async context manager."""
394 if self._client is not None:
395 await self._client.aclose()
396 self._client = None
398 async def request(
399 self,
400 method: str,
401 url: str,
402 **kwargs: Unpack[HttpRequestKwargs],
403 ) -> Result[httpx.Response, Exception]:
404 """Send an HTTP request with safety features.
406 Args:
407 method: HTTP method.
408 url: Target URL.
409 **kwargs: Passed to the underlying client.
411 Returns:
412 ``Ok(Response)`` on success, ``Err`` on failure.
414 """
415 if self._client is None:
416 return Err(RuntimeError("Client not initialised. Use 'async with'."))
418 precondition_err = _check_preconditions(
419 url, self._ssrf_protection, self._circuit_breaker
420 )
421 if precondition_err is not None:
422 return precondition_err
424 async def _do_request() -> httpx.Response:
425 # We explicitly verified client is not None above
426 client: httpx.AsyncClient = cast(httpx.AsyncClient, self._client)
427 request_func = cast(
428 Callable[..., Awaitable[httpx.Response]],
429 client.request,
430 )
431 response = await request_func(method, url, **kwargs)
432 return response
434 return await _execute_with_retries(
435 _do_request,
436 self._retry_config,
437 self._circuit_breaker,
438 self._retryable_status_codes,
439 )
441 async def get(
442 self,
443 url: str,
444 **kw: Unpack[HttpRequestKwargs],
445 ) -> Result[httpx.Response, Exception]:
446 """Send a GET request."""
447 return await self.request("GET", url, **kw)
449 async def post(
450 self,
451 url: str,
452 **kw: Unpack[HttpRequestKwargs],
453 ) -> Result[httpx.Response, Exception]:
454 """Send a POST request."""
455 return await self.request("POST", url, **kw)
457 async def put(
458 self,
459 url: str,
460 **kw: Unpack[HttpRequestKwargs],
461 ) -> Result[httpx.Response, Exception]:
462 """Send a PUT request."""
463 return await self.request("PUT", url, **kw)
465 async def delete(
466 self,
467 url: str,
468 **kw: Unpack[HttpRequestKwargs],
469 ) -> Result[httpx.Response, Exception]:
470 """Send a DELETE request."""
471 return await self.request("DELETE", url, **kw)
473 async def patch(
474 self,
475 url: str,
476 **kw: Unpack[HttpRequestKwargs],
477 ) -> Result[httpx.Response, Exception]:
478 """Send a PATCH request."""
479 return await self.request("PATCH", url, **kw)