Skip to content

Retry

The RetryPolicy dataclass configures how the client retries transient failures (429, 5xx, network errors). It is a frozen, slotted dataclass with test-friendly class methods.

retry

HTTP_TOO_MANY module-attribute 

HTTP_TOO_MANY = 429

HTTP_SERVER_MIN module-attribute 

HTTP_SERVER_MIN = 500

HTTP_SERVER_MAX module-attribute 

HTTP_SERVER_MAX = 600

__all__ module-attribute 

__all__ = ['RetryPolicy', 'is_retryable']

RetryPolicy dataclass 

RetryPolicy(
    max_attempts: int = 6,
    initial_seconds: float = 1.0,
    max_seconds: float = 60.0,
    jitter_seconds: float = 2.0,
)

How the Forecast client handles transient failures.

max_attempts class-attribute instance-attribute 

max_attempts: int = 6

initial_seconds class-attribute instance-attribute 

initial_seconds: float = 1.0

max_seconds class-attribute instance-attribute 

max_seconds: float = 60.0

jitter_seconds class-attribute instance-attribute 

jitter_seconds: float = 2.0

no_wait classmethod 

no_wait() -> RetryPolicy

Test-friendly: zero backoff, single attempt.

Source code in src/harvest_forecast/retry.py 
28
29
30
31
@classmethod
def no_wait(cls) -> RetryPolicy:
    """Test-friendly: zero backoff, single attempt."""
    return cls(max_attempts=1, initial_seconds=0.0, max_seconds=0.0, jitter_seconds=0.0)

fast_test classmethod 

fast_test(max_attempts: int = 3) -> RetryPolicy

Test-friendly: zero backoff, configurable attempt count.

Source code in src/harvest_forecast/retry.py 
33
34
35
36
37
38
39
40
41
@classmethod
def fast_test(cls, max_attempts: int = 3) -> RetryPolicy:
    """Test-friendly: zero backoff, configurable attempt count."""
    return cls(
        max_attempts=max_attempts,
        initial_seconds=0.0,
        max_seconds=0.0,
        jitter_seconds=0.0,
    )

wait 

wait(state: RetryCallState) -> float
Source code in src/harvest_forecast/retry.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
def wait(self, state: RetryCallState) -> float:
    if self.initial_seconds == 0.0 and self.max_seconds == 0.0:
        base = 0.0
    else:
        base = wait_exponential_jitter(
            initial=self.initial_seconds,
            max=self.max_seconds,
            jitter=self.jitter_seconds,
        )(state)
    exc = state.outcome.exception() if state.outcome else None
    if isinstance(exc, ForecastRateLimitError) and exc.retry_after is not None:
        return max(base, exc.retry_after) if self.max_seconds > 0 else 0.0
    return base

is_retryable 

is_retryable(exc: BaseException) -> bool
Source code in src/harvest_forecast/retry.py 
15
16
def is_retryable(exc: BaseException) -> bool:
    return isinstance(exc, httpx.TransportError | ForecastRateLimitError | ForecastServerError)