API

class RetryTransport(httpx.BaseTransport, httpx.AsyncBaseTransport):
    """
    A transport that automatically retries requests.

    ```python
    with httpx.Client(transport=RetryTransport()) as client:
        response = client.get("https://example.com")

    async with httpx.AsyncClient(transport=RetryTransport()) as client:
        response = await client.get("https://example.com")
    ```

    If you want to use a specific retry strategy, provide a [Retry][httpx_retries.Retry] configuration:

    ```python
    retry = Retry(total=5, backoff_factor=0.5)
    transport = RetryTransport(retry=retry)

    with httpx.Client(transport=transport) as client:
        response = client.get("https://example.com")
    ```

    By default, the implementation will create a sync and async transport internally, and use whichever is appropriate
    for the request. If you want to configure your own transport, provide it to the `transport` argument:

    ```python
    transport = RetryTransport(transport=httpx.HTTPTransport(local_address="0.0.0.0"))
    ```

    Args:
        transport: Optional transport to wrap. If not provided, async and sync transports are created internally.
        retry: The retry configuration.
    """

    def __init__(
        self,
        transport: httpx.BaseTransport | httpx.AsyncBaseTransport | None = None,
        retry: Retry | None = None,
    ) -> None:
        self.retry = retry or Retry()

        if transport is not None:
            self._sync_transport = transport if isinstance(transport, httpx.BaseTransport) else None
            self._async_transport = transport if isinstance(transport, httpx.AsyncBaseTransport) else None
        else:
            self._sync_transport = httpx.HTTPTransport()
            self._async_transport = httpx.AsyncHTTPTransport()

    def close(self) -> None:
        """
        Closes this transport.
        """
        if self._sync_transport is not None:
            self._sync_transport.close()

    async def aclose(self) -> None:
        """
        Closes this transport.
        """
        if self._async_transport is not None:
            await self._async_transport.aclose()

    def handle_request(self, request: httpx.Request) -> httpx.Response:
        """
        Sends an HTTP request, possibly with retries.

        Args:
            request (httpx.Request): The request to send.

        Returns:
            The final response.
        """
        if self._sync_transport is None:
            raise RuntimeError("Synchronous request received but no sync transport available")

        logger.debug("handle_request started request=%s", request)

        retry: Retry = request.extensions.setdefault("retry", self.retry)

        if retry.is_retryable_method(request.method):
            if retry.validate_response is not None and inspect.iscoroutinefunction(retry.validate_response):
                raise TypeError("validate_response must be a sync function when using a sync transport")

            send_method = partial(self._sync_transport.handle_request)
            response = _retry_operation(request, send_method, retry)
        else:
            response = self._sync_transport.handle_request(request)

        logger.debug("handle_request finished request=%s response=%s", request, response)

        return response

    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        """Sends an HTTP request, possibly with retries.

        Args:
            request: The request to perform.

        Returns:
            The final response.
        """
        if self._async_transport is None:
            raise RuntimeError("Async request received but no async transport available")

        logger.debug("handle_async_request started request=%s", request)

        retry: Retry = request.extensions.setdefault("retry", self.retry)

        if retry.is_retryable_method(request.method):
            send_method = partial(self._async_transport.handle_async_request)
            response = await _retry_operation_async(request, send_method, retry)
        else:
            response = await self._async_transport.handle_async_request(request)

        logger.debug("handle_async_request finished request=%s response=%s", request, response)

        return response

class Retry:
    """
    A class to encapsulate retry logic and configuration.

    Each retry attempt will create a new [Retry][httpx_retries.Retry] object with updated values,
    so they can safely be reused.

    If `backoff_factor` is set, it will use an exponential backoff with configurable jitter.

    For complex use cases, you can override the `backoff_strategy` method.

    Args:
        total (int, optional): The maximum number of times to retry a request before giving up.
        max_backoff_wait (float, optional): The maximum time in seconds to wait between retries.
        backoff_factor (float, optional): The factor by which the wait time increases with each retry attempt.
        respect_retry_after_header (bool, optional): Whether to respect the Retry-After header in HTTP responses
            when deciding how long to wait before retrying.
        allowed_methods (Iterable[http.HTTPMethod, str], optional): The HTTP methods that can be retried. Defaults to
            ["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"].
        status_forcelist (Iterable[http.HTTPStatus, int], optional): The HTTP status codes that can be retried.
            Defaults to [429, 502, 503, 504].
        retry_on_exceptions (Iterable[type[httpx.HTTPError]], optional): The HTTP exceptions that can be retried.
            Defaults to [httpx.TimeoutException, httpx.NetworkError, httpx.RemoteProtocolError].
        backoff_jitter (float, optional): The amount of jitter to add to the backoff time, between 0 and 1.
            Defaults to 1 (full jitter).
        attempts_made (int, optional): The number of retry attempts already made.
        total_timeout (float, optional): The maximum cumulative time in seconds to spend sleeping between retry
            attempts across a single request. Unlike `max_backoff_wait` (which caps a single sleep), this caps the
            sum of all sleeps. Useful as a defence against a server that returns a large `Retry-After`
            repeatedly. Defaults to None (no cumulative cap).
        elapsed_sleep (float, optional): Cumulative sleep time already spent on this request. Preserved across
            `increment()` calls; users typically do not set this directly.
        validate_response (callable, optional): An optional callback called with each response that would
            otherwise be returned as a "good" (non-retryable-status) response. If the callback raises, the
            request is retried. May be sync or async; an async callback cannot be used with a sync transport.
            Signature: ``(response: httpx.Response) -> None``.
    """

    RETRYABLE_METHODS: Final[frozenset[HTTPMethod]] = frozenset(
        [
            HTTPMethod.HEAD,
            HTTPMethod.GET,
            HTTPMethod.PUT,
            HTTPMethod.DELETE,
            HTTPMethod.OPTIONS,
            HTTPMethod.TRACE,
        ]
    )
    RETRYABLE_STATUS_CODES: Final[frozenset[HTTPStatus]] = frozenset(
        [
            HTTPStatus.TOO_MANY_REQUESTS,
            HTTPStatus.BAD_GATEWAY,
            HTTPStatus.SERVICE_UNAVAILABLE,
            HTTPStatus.GATEWAY_TIMEOUT,
        ]
    )
    RETRYABLE_EXCEPTIONS: Final[tuple[type[Exception], ...]] = (
        httpx.TimeoutException,
        httpx.NetworkError,
        httpx.RemoteProtocolError,
    )

    def __init__(
        self,
        total: int = 10,
        allowed_methods: Iterable[HTTPMethod | str] | None = None,
        status_forcelist: Iterable[HTTPStatus | int] | None = None,
        retry_on_exceptions: Iterable[type[Exception]] | None = None,
        backoff_factor: float = 0.0,
        respect_retry_after_header: bool = True,
        max_backoff_wait: float = 120.0,
        backoff_jitter: float = 1.0,
        attempts_made: int = 0,
        total_timeout: float | None = None,
        elapsed_sleep: float = 0.0,
        validate_response: Callable[[httpx.Response], None | Awaitable[None]] | None = None,
    ) -> None:
        """Initialize a new Retry instance."""
        if total < 0:
            raise ValueError("total must be non-negative")
        if backoff_factor < 0:
            raise ValueError("backoff_factor must be non-negative")
        if max_backoff_wait <= 0:
            raise ValueError("max_backoff_wait must be positive")
        if not 0 <= backoff_jitter <= 1:
            raise ValueError("backoff_jitter must be between 0 and 1")
        if attempts_made < 0:
            raise ValueError("attempts_made must be non-negative")
        if total_timeout is not None and total_timeout <= 0:
            raise ValueError("total_timeout must be positive")
        if elapsed_sleep < 0:
            raise ValueError("elapsed_sleep must be non-negative")

        self.total = total
        self.backoff_factor = backoff_factor
        self.respect_retry_after_header = respect_retry_after_header
        self.max_backoff_wait = max_backoff_wait
        self.backoff_jitter = backoff_jitter
        self.attempts_made = attempts_made
        self.total_timeout = total_timeout
        self.elapsed_sleep = elapsed_sleep
        self.validate_response = validate_response

        self.allowed_methods: frozenset[str] = frozenset(
            method.upper() for method in (allowed_methods or self.RETRYABLE_METHODS)
        )
        self.status_forcelist = frozenset((status_forcelist or self.RETRYABLE_STATUS_CODES))
        self.retryable_exceptions = (
            self.RETRYABLE_EXCEPTIONS if retry_on_exceptions is None else tuple(retry_on_exceptions)
        )

    def is_retryable_method(self, method: str) -> bool:
        """Check if a method is retryable."""
        return method.upper() in self.allowed_methods

    def is_retryable_status_code(self, status_code: int) -> bool:
        """Check if a status code is retryable."""
        return status_code in self.status_forcelist

    def is_retryable_exception(self, exception: Exception) -> bool:
        """Check if an exception is retryable."""
        return isinstance(exception, self.retryable_exceptions)

    def is_retry(self, method: str, status_code: int, has_retry_after: bool) -> bool:
        """
        Check if a method and status code are retryable.

        This functions identically to urllib3's `Retry.is_retry` method.
        """
        return (
            self.total > 0
            and self.is_retryable_method(method)
            and self.is_retryable_status_code(status_code)
            and not has_retry_after
        )

    def is_exhausted(self) -> bool:
        """Check if the retry attempts have been exhausted."""
        if self.attempts_made >= self.total:
            return True
        if self.total_timeout is not None and self.elapsed_sleep >= self.total_timeout:
            return True
        return False

    def parse_retry_after(self, retry_after: str) -> float:
        """
        Parse the Retry-After header.

        Args:
            retry_after: The Retry-After header value.

        Returns:
            The number of seconds to wait before retrying.

        Raises:
            ValueError: If the Retry-After header is not a valid number or HTTP date.
        """
        retry_after = retry_after.strip()
        if retry_after.isascii() and retry_after.isdigit():
            return float(retry_after)

        try:
            parsed_date = parsedate_to_datetime(retry_after)
            if parsed_date.tzinfo is None:
                logger.warning("Retry-After date has no timezone info, assuming UTC: %s", retry_after)
                parsed_date = parsed_date.replace(tzinfo=datetime.timezone.utc)

            diff = (parsed_date - datetime.datetime.now(datetime.timezone.utc)).total_seconds()
            return max(0.0, diff)
        except (TypeError, ValueError):
            raise ValueError(f"Invalid Retry-After header: {retry_after}")

    def backoff_strategy(self) -> float:
        """
        Calculate the backoff time based on the number of attempts.

        For complex use cases, you can override this method to implement a custom backoff strategy.

        ```python
        class CustomRetry(Retry):
            def backoff_strategy(self) -> float:
                if self.attempts_made == 3:
                    return 1.0

                return super().backoff_strategy()
        ```

        Returns:
            The calculated backoff time in seconds, capped by max_backoff_wait.
        """
        if self.backoff_factor == 0:
            return 0.0

        # Calculate exponential backoff
        backoff: float = self.backoff_factor * (2**self.attempts_made)

        # Apply jitter if configured
        if self.backoff_jitter > 0:
            backoff *= random.uniform(1 - self.backoff_jitter, 1)

        return min(backoff, self.max_backoff_wait)

    def _calculate_sleep(self, headers: httpx.Headers | Mapping[str, str]) -> float:
        """Calculate the sleep duration based on headers and backoff strategy."""
        sleep_time = 0.0
        # Check Retry-After header first if enabled
        if self.respect_retry_after_header:
            retry_after = headers.get("Retry-After", "").strip()
            if retry_after:
                try:
                    retry_after_sleep = min(self.parse_retry_after(retry_after), self.max_backoff_wait)
                    if retry_after_sleep > 0:
                        sleep_time = retry_after_sleep
                except ValueError:
                    logger.warning("Retry-After header is not a valid HTTP date: %s", retry_after)

        # Fall back to backoff strategy
        if sleep_time == 0.0:
            sleep_time = self.backoff_strategy() if self.attempts_made > 0 else 0.0

        # Cap by remaining total_timeout budget to bound cumulative sleep
        if self.total_timeout is not None:
            remaining = max(0.0, self.total_timeout - self.elapsed_sleep)
            sleep_time = min(sleep_time, remaining)

        return sleep_time

    def sleep(self, response: httpx.Response | Exception) -> None:
        """
        Sleep between retry attempts using the calculated duration.

        This method will respect a server’s `Retry-After` response header and sleep the duration
        of the time requested. If that is not present, it will use an exponential backoff. By default,
        the backoff factor is 0 and this method will return immediately.
        """
        time_to_sleep = self._calculate_sleep(response.headers if isinstance(response, httpx.Response) else {})
        logger.debug("sleep seconds=%s", time_to_sleep)
        time.sleep(time_to_sleep)
        self.elapsed_sleep += time_to_sleep

    async def asleep(self, response: httpx.Response | Exception) -> None:
        """
        Sleep between retry attempts asynchronously using the calculated duration.

        This method will respect a server’s `Retry-After` response header and sleep the duration
        of the time requested. If that is not present, it will use an exponential backoff. By default,
        the backoff factor is 0 and this method will return immediately.
        """
        time_to_sleep = self._calculate_sleep(response.headers if isinstance(response, httpx.Response) else {})
        logger.debug("asleep seconds=%s", time_to_sleep)
        await asyncio.sleep(time_to_sleep)
        self.elapsed_sleep += time_to_sleep

    def copy_with(
        self,
        total: int | _UnsetType = _UNSET,
        allowed_methods: Iterable[HTTPMethod | str] | None | _UnsetType = _UNSET,
        status_forcelist: Iterable[HTTPStatus | int] | None | _UnsetType = _UNSET,
        retry_on_exceptions: Iterable[type[Exception]] | None | _UnsetType = _UNSET,
        backoff_factor: float | _UnsetType = _UNSET,
        respect_retry_after_header: bool | _UnsetType = _UNSET,
        max_backoff_wait: float | _UnsetType = _UNSET,
        backoff_jitter: float | _UnsetType = _UNSET,
        attempts_made: int | _UnsetType = _UNSET,
        total_timeout: float | None | _UnsetType = _UNSET,
        elapsed_sleep: float | _UnsetType = _UNSET,
        validate_response: Callable[[httpx.Response], None | Awaitable[None]] | None | _UnsetType = _UNSET,
    ) -> "Retry":
        """Return a new Retry with selected fields overridden."""
        return self.__class__(
            total=self.total if isinstance(total, _UnsetType) else total,
            allowed_methods=self.allowed_methods if isinstance(allowed_methods, _UnsetType) else allowed_methods,
            status_forcelist=self.status_forcelist if isinstance(status_forcelist, _UnsetType) else status_forcelist,
            retry_on_exceptions=self.retryable_exceptions
            if isinstance(retry_on_exceptions, _UnsetType)
            else retry_on_exceptions,
            backoff_factor=self.backoff_factor if isinstance(backoff_factor, _UnsetType) else backoff_factor,
            respect_retry_after_header=self.respect_retry_after_header
            if isinstance(respect_retry_after_header, _UnsetType)
            else respect_retry_after_header,
            max_backoff_wait=self.max_backoff_wait if isinstance(max_backoff_wait, _UnsetType) else max_backoff_wait,
            backoff_jitter=self.backoff_jitter if isinstance(backoff_jitter, _UnsetType) else backoff_jitter,
            attempts_made=self.attempts_made if isinstance(attempts_made, _UnsetType) else attempts_made,
            total_timeout=self.total_timeout if isinstance(total_timeout, _UnsetType) else total_timeout,
            elapsed_sleep=self.elapsed_sleep if isinstance(elapsed_sleep, _UnsetType) else elapsed_sleep,
            validate_response=self.validate_response
            if isinstance(validate_response, _UnsetType)
            else validate_response,
        )

    def increment(self) -> "Retry":
        """Return a new Retry instance with the attempt count incremented."""
        logger.debug("increment retry=%s new_attempts_made=%s", self, self.attempts_made + 1)
        return self.copy_with(attempts_made=self.attempts_made + 1)

    def __repr__(self) -> str:
        return f"<Retry(total={self.total}, attempts_made={self.attempts_made})>"