Retry policy

RetryPolicy opts the client into transient-failure retries. Default-off — pass retry_policy=RetryPolicy(...) to Dhis2Client and the underlying httpx transport gets wrapped with exponential backoff, jitter, and Retry-After support.

Only idempotent methods (GET, HEAD, PUT, DELETE, OPTIONS) retry by default. Set retry_non_idempotent=True to also retry POST and PATCH — leave it off unless the specific endpoint is known to be idempotent.

retry

Retry policy + httpx transport wrapper for transient HTTP failures.

Default-off. Opt in by passing retry_policy=RetryPolicy(...) to Dhis2Client — the client wraps its httpx transport with _RetryTransport which retries on the configured status codes and on connection-level errors (ConnectError, ReadTimeout, etc.) with exponential backoff + jitter.

Design:

• Only idempotent HTTP methods (GET, HEAD, PUT, DELETE, OPTIONS) retry by default. POST / PATCH are skipped unless retry_non_idempotent=True — double-writes can cause DHIS2-side duplicates (duplicate create, double delete, etc). Caller opts in explicitly when they know the endpoint is safe (analytics refresh kick-offs, for instance).
• A server-provided Retry-After header (sent on 429 / 503) overrides the computed backoff for that attempt.
• Exponential: delay = min(max_delay, base_delay * backoff_factor ** (attempt - 1)) with a ± jitter applied before sleeping.

Classes

RetryPolicy

Bases: BaseModel

Retry config for transient HTTP failures.

Source code in packages/dhis2w-client/src/dhis2w_client/retry.py

class RetryPolicy(BaseModel):
    """Retry config for transient HTTP failures."""

    model_config = ConfigDict(frozen=True)

    max_attempts: int = Field(default=3, ge=1, le=10, description="Total attempt count including the first call.")
    base_delay: float = Field(default=0.5, ge=0.0, description="Initial backoff in seconds before the second attempt.")
    max_delay: float = Field(default=30.0, gt=0.0, description="Hard cap on the per-sleep delay.")
    backoff_factor: float = Field(default=2.0, ge=1.0, description="Multiplier applied per attempt (exponential).")
    jitter: float = Field(
        default=0.1,
        ge=0.0,
        le=1.0,
        description="Fractional random jitter applied to each delay (0.1 = +/- 10%).",
    )
    retry_statuses: frozenset[int] = Field(
        default=frozenset({429, 502, 503, 504}),
        description="HTTP status codes that trigger a retry. Defaults match transient-failure conventions.",
    )
    retry_non_idempotent: bool = Field(
        default=False,
        description="When True, retry POST/PATCH too. Leave False unless you know the endpoint is idempotent.",
    )

    def compute_delay(self, attempt: int, *, rng: random.Random | None = None) -> float:
        """Compute the sleep before attempt `attempt` (1-based; `compute_delay(1)` is before the 2nd try)."""
        raw = self.base_delay * (self.backoff_factor ** max(0, attempt - 1))
        capped = min(self.max_delay, raw)
        if self.jitter <= 0.0:
            return capped
        rnd = rng or random
        # Sample from [1 - jitter, 1 + jitter] so it's symmetric around the nominal delay.
        return capped * (1.0 + rnd.uniform(-self.jitter, self.jitter))

Functions

compute_delay(attempt, *, rng=None)

Compute the sleep before attempt attempt (1-based; compute_delay(1) is before the 2^nd try).

Source code in packages/dhis2w-client/src/dhis2w_client/retry.py

def compute_delay(self, attempt: int, *, rng: random.Random | None = None) -> float:
    """Compute the sleep before attempt `attempt` (1-based; `compute_delay(1)` is before the 2nd try)."""
    raw = self.base_delay * (self.backoff_factor ** max(0, attempt - 1))
    capped = min(self.max_delay, raw)
    if self.jitter <= 0.0:
        return capped
    rnd = rng or random
    # Sample from [1 - jitter, 1 + jitter] so it's symmetric around the nominal delay.
    return capped * (1.0 + rnd.uniform(-self.jitter, self.jitter))

Functions

build_retry_transport(policy, *, inner=None)

Compose a retry-wrapped transport; defaults to wrapping a fresh AsyncHTTPTransport.

Source code in packages/dhis2w-client/src/dhis2w_client/retry.py

def build_retry_transport(policy: RetryPolicy, *, inner: httpx.AsyncBaseTransport | None = None) -> _RetryTransport:
    """Compose a retry-wrapped transport; defaults to wrapping a fresh `AsyncHTTPTransport`."""
    base: httpx.AsyncBaseTransport = inner if inner is not None else _default_transport()
    return _RetryTransport(base, policy)