HTTPX Async Client Retry Transport

import asyncio
import random

import httpx


class RetryingTransport(httpx.AsyncHTTPTransport):
    def __init__(self, retries: int = 3, **kwargs):
        super().__init__(**kwargs)
        self._retries = retries

    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        retryable = request.method in {'GET', 'HEAD', 'OPTIONS'}
        last_error: Exception | None = None

        for attempt in range(self._retries + 1):
            if attempt > 0:
                backoff = min(2 ** attempt * 0.5, 8.0)
                await asyncio.sleep(backoff + random.uniform(0, 0.3))
            try:
                response = await super().handle_async_request(request)
                if response.status_code in {429, 502, 503, 504} and retryable:
                    retry_after = response.headers.get('Retry-After')
                    if retry_after and retry_after.isdigit():
                        await asyncio.sleep(int(retry_after))
                    continue
                return response
            except (httpx.TransportError,) as exc:
                last_error = exc
        raise httpx.TransportError(f'exhausted retries: {last_error}')


async def main() -> None:
    transport = RetryingTransport(retries=3)
    async with httpx.AsyncClient(transport=transport, timeout=10.0) as client:
        resp = await client.get('https://api.example.com/status')
        print('final status:', resp.status_code)


asyncio.run(main())