Retry & Transports Reference

HTTP transport middleware with automatic exponential backoff and jitter.

RetryTransport

RetryTransport(transport: BaseTransport, max_retries: int = DEFAULT_MAX_RETRIES, backoff_factor: float = DEFAULT_BACKOFF_FACTOR)

Bases: BaseTransport

Synchronous HTTP transport middleware with exponential backoff and jitter.

Retries transient 5xx server errors, connection timeouts, and short-duration transient 429 rate limit responses.

Initialize the retry transport wrapper.

Parameters:

Name	Type	Description	Default
transport	BaseTransport	The underlying HTTP transport to delegate requests to.	required
max_retries	int	Maximum number of retry attempts before raising.	DEFAULT_MAX_RETRIES
backoff_factor	float	Multiplier for exponential backoff calculations.	DEFAULT_BACKOFF_FACTOR

Source code in src/cotbias/retry.py

def __init__(
    self,
    transport: httpx.BaseTransport,
    max_retries: int = DEFAULT_MAX_RETRIES,
    backoff_factor: float = DEFAULT_BACKOFF_FACTOR,
) -> None:
    """Initialize the retry transport wrapper.

    Args:
        transport: The underlying HTTP transport to delegate requests to.
        max_retries: Maximum number of retry attempts before raising.
        backoff_factor: Multiplier for exponential backoff calculations.
    """
    self._inner = transport
    self._max_retries = max(0, max_retries)
    self._backoff = backoff_factor

handle_request

handle_request(request: Request) -> httpx.Response

Send the request through the inner transport with automatic retry on transient failure.

Parameters:

Name	Type	Description	Default
request	Request	The outgoing HTTP request.	required

Returns:

Type	Description
Response	The HTTP response received from the server.

Source code in src/cotbias/retry.py

def handle_request(self, request: httpx.Request) -> httpx.Response:
    """Send the request through the inner transport with automatic retry on transient failure.

    Args:
        request: The outgoing HTTP request.

    Returns:
        The HTTP response received from the server.
    """
    request.read()
    attempts = 0

    while True:
        try:
            resp = self._inner.handle_request(request)
            if resp.status_code >= 400:
                resp.read()
            if attempts < self._max_retries:
                retry, delay = _should_retry_response(resp)
                if retry:
                    attempts += 1
                    sleep_time = (
                        delay
                        if delay is not None
                        else self._backoff * (2 ** (attempts - 1)) + random.uniform(0, 0.1)
                    )
                    logger.warning(
                        "Transient %d received for %s. Retrying in %.2fs (attempt %d/%d)...",
                        resp.status_code,
                        request.url,
                        sleep_time,
                        attempts,
                        self._max_retries,
                    )
                    time.sleep(sleep_time)
                    continue
            return resp
        except (httpx.TimeoutException, httpx.NetworkError) as e:
            if attempts < self._max_retries:
                attempts += 1
                sleep_time = self._backoff * (2 ** (attempts - 1)) + random.uniform(0, 0.1)
                logger.warning(
                    "Network/timeout error (%s) on %s. Retrying in %.2fs (attempt %d/%d)...",
                    type(e).__name__,
                    request.url,
                    sleep_time,
                    attempts,
                    self._max_retries,
                )
                time.sleep(sleep_time)
                continue
            raise

close

close() -> None

Close the underlying inner transport.

Source code in src/cotbias/retry.py

def close(self) -> None:
    """Close the underlying inner transport."""
    self._inner.close()

AsyncRetryTransport

AsyncRetryTransport(transport: AsyncBaseTransport, max_retries: int = DEFAULT_MAX_RETRIES, backoff_factor: float = DEFAULT_BACKOFF_FACTOR)

Bases: AsyncBaseTransport

Asynchronous HTTP transport middleware with exponential backoff and jitter.

Retries transient 5xx server errors, connection timeouts, and short-duration transient 429 rate limit responses asynchronously.

Initialize the async retry transport wrapper.

Parameters:

Name	Type	Description	Default
transport	AsyncBaseTransport	The underlying async HTTP transport to delegate requests to.	required
max_retries	int	Maximum number of retry attempts before raising.	DEFAULT_MAX_RETRIES
backoff_factor	float	Multiplier for exponential backoff calculations.	DEFAULT_BACKOFF_FACTOR

Source code in src/cotbias/retry.py

def __init__(
    self,
    transport: httpx.AsyncBaseTransport,
    max_retries: int = DEFAULT_MAX_RETRIES,
    backoff_factor: float = DEFAULT_BACKOFF_FACTOR,
) -> None:
    """Initialize the async retry transport wrapper.

    Args:
        transport: The underlying async HTTP transport to delegate requests to.
        max_retries: Maximum number of retry attempts before raising.
        backoff_factor: Multiplier for exponential backoff calculations.
    """
    self._inner = transport
    self._max_retries = max(0, max_retries)
    self._backoff = backoff_factor

handle_async_request async

handle_async_request(request: Request) -> httpx.Response

Send the async request through the inner transport with automatic retry on transient failure.

Parameters:

Name	Type	Description	Default
request	Request	The outgoing HTTP request.	required

Returns:

Type	Description
Response	The HTTP response received from the server.

Source code in src/cotbias/retry.py

async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
    """Send the async request through the inner transport with automatic retry on transient failure.

    Args:
        request: The outgoing HTTP request.

    Returns:
        The HTTP response received from the server.
    """
    await request.aread()
    attempts = 0

    while True:
        try:
            resp = await self._inner.handle_async_request(request)
            if resp.status_code >= 400:
                await resp.aread()
            if attempts < self._max_retries:
                retry, delay = _should_retry_response(resp)
                if retry:
                    attempts += 1
                    sleep_time = (
                        delay
                        if delay is not None
                        else self._backoff * (2 ** (attempts - 1)) + random.uniform(0, 0.1)
                    )
                    logger.warning(
                        "Transient %d received for %s. Retrying in %.2fs (attempt %d/%d)...",
                        resp.status_code,
                        request.url,
                        sleep_time,
                        attempts,
                        self._max_retries,
                    )
                    await asyncio.sleep(sleep_time)
                    continue
            return resp
        except (httpx.TimeoutException, httpx.NetworkError) as e:
            if attempts < self._max_retries:
                attempts += 1
                sleep_time = self._backoff * (2 ** (attempts - 1)) + random.uniform(0, 0.1)
                logger.warning(
                    "Network/timeout error (%s) on %s. Retrying in %.2fs (attempt %d/%d)...",
                    type(e).__name__,
                    request.url,
                    sleep_time,
                    attempts,
                    self._max_retries,
                )
                await asyncio.sleep(sleep_time)
                continue
            raise

aclose async

aclose() -> None

Close the underlying inner async transport.

Source code in src/cotbias/retry.py

async def aclose(self) -> None:
    """Close the underlying inner async transport."""
    await self._inner.aclose()