RFC: base retry class for building custom retry rules

[mharrisb1]

Request for comment (RFC) for potential feature:

To allow a user to define more complex retries not available through constructor options we could provide a retry base class that has abstract methods that the user can inherit from and design.

For example, if I want infinite retries on 1XX status codes with not backoff or jitter but provide an allowance for 4XX and 5XX status codes with backoff or jitter I don't see a way to do that today.

Example:

from httpx_retries import BaseRetry, RetryTransport

class MySuperComplexRetry(BaseRetry):
    # need to clarify which abstract methods need to be defined
    ...

transport = RetryTransport(retry=MySuperComplexRetry())

An alternative already exists for this today since the user can technically just subclass Retry and override methods but this may present a cleaner option so that we could define the exact methods that an BaseRetry child class must define.

[will-ockmore]

I think this is a valid problem, and that we should solve it by improving what's available to a user in the Retry class. Adding and additional base class is interesting, but we can probably avoid it.

One thing I'd like to add to Retry that would make your example possible to implement just by overriding backoff_strategy is response history. Calls to Retry.sleep should add to a sequence of responses (or errors) that have caused a retry to occur; then this state could be inspected to drive complex behaviour.

For your example, you could do something like

class CustomRetry(Retry):
    def backoff_strategy(self) -> float:
        last_response = self.history[-1]

        if isinstance(last_response, httpx.Response) and 100 <= last_response.status_code < 200:
            return 1.0

        return super().backoff_strategy()

[devjerry0]

Here is a simple example I used to log retries.

class LoggedRetry(Retry):
    """Retry that emits structured warnings and span events before each back-off sleep."""

    async def asleep(self, response: httpx.Response | Exception) -> None:
        delay = self._calculate_sleep(
            response.headers if isinstance(response, httpx.Response) else {},
        )
        log_attrs = {
            "retry.attempt": self.attempts_made,
            "retry.total": self.total,
            "retry.delay": round(delay, 3),
            "retry.error": str(response),
        }
        _log.warning("retrying_request", **log_attrs)
        span_attrs: dict[str, str | bool | int | float] = {
            "retry.attempt": self.attempts_made,
            "retry.total": self.total,
            "retry.delay": round(delay, 3),
            "retry.error": str(response),
        }
        trace.get_current_span().add_event("retry", attributes=span_attrs)
        await super().asleep(response)