Retry Strategies API Reference

This page documents retry strategy implementations and interfaces.

RetryStrategyInterface

class RetryStrategyInterface

Abstract interface for retry strategies.

All retry strategies must implement this interface.

Methods

RetryStrategyInterface.should_retry(context, exception)

Determine if request should be retried.

Parameters:

context (RequestContext) – Request context with attempt count
exception (Exception) – Exception that caused failure

Returns:

True if should retry, False otherwise

Return type:

bool

RetryStrategyInterface.get_delay(context)

Get delay in seconds before next retry.

Parameters:: context (RequestContext) – Request context
Returns:: Delay in seconds
Return type:: float

Properties

max_retries: int: Maximum number of retry attempts.

NoRetryStrategy

class NoRetryStrategy

No retry strategy - fail immediately on error.

Returns:: NoRetryStrategy instance
Return type:: NoRetryStrategy

Use Cases:

Unit tests requiring immediate failures
Non-critical requests where retry overhead isn’t justified
Operations that should never be retried

Example:

from requestforge import NoRetryStrategy

strategy = NoRetryStrategy()

print(strategy.max_retries)  # 0
print(strategy.should_retry(context, exception))  # Always False
print(strategy.get_delay(context))  # 0.0

SimpleRetryStrategy

class SimpleRetryStrategy(max_retries=3, delay=1.0, retryable_exceptions=None)

Simple retry strategy with fixed delay.

Parameters:

max_retries (int) – Maximum retry attempts
delay (float) – Fixed delay between retries in seconds
retryable_exceptions (frozenset) – Exception types to retry (optional)

Attributes

SimpleRetryStrategy.max_retries: int: Maximum number of retry attempts.

Methods

SimpleRetryStrategy.should_retry(context, exception)

Check if should retry based on attempt count and exception type.

Parameters:

context (RequestContext) – Request context
exception (Exception) – Exception that occurred

Returns:

True if should retry

Return type:

bool

SimpleRetryStrategy.get_delay(context)

Get fixed delay.

Parameters:: context (RequestContext) – Request context
Returns:: Fixed delay in seconds
Return type:: float

Examples

from requestforge import SimpleRetryStrategy, TimeoutException, ConnectionException

# Basic usage
strategy = SimpleRetryStrategy(max_retries=3, delay=2.0)

# Only retry specific exceptions
strategy = SimpleRetryStrategy(
    max_retries=3,
    delay=1.0,
    retryable_exceptions=frozenset({
        TimeoutException,
        ConnectionException
    })
)

# Configure client
config = builder.with_retry_strategy(strategy).build()

ExponentialBackoffRetryStrategy

class ExponentialBackoffRetryStrategy(max_retries=3, base_delay=1.0, max_delay=60.0, multiplier=2.0, jitter=True, retryable_exceptions=None, retryable_status_codes=None)

Exponential backoff retry strategy with jitter.

Recommended for production use.

Parameters:

max_retries (int) – Maximum retry attempts
base_delay (float) – Initial delay in seconds
max_delay (float) – Maximum delay cap in seconds
multiplier (float) – Exponential multiplier
jitter (bool) – Add random jitter (±25%)
retryable_exceptions (frozenset) – Exception types to retry
retryable_status_codes (frozenset) – HTTP status codes to retry

Attributes

ExponentialBackoffRetryStrategy.max_retries: int: Maximum number of retry attempts.

Methods

ExponentialBackoffRetryStrategy.should_retry(context, exception)

Check if should retry based on attempt count, exception type, and status code.

Parameters:

context (RequestContext) – Request context
exception (Exception) – Exception that occurred

Returns:

True if should retry

Return type:

bool

ExponentialBackoffRetryStrategy.get_delay(context)

Calculate exponential backoff delay with optional jitter.

Parameters:: context (RequestContext) – Request context
Returns:: Delay in seconds
Return type:: float

Delay Formula:

delay = min(base_delay * (multiplier ** attempt), max_delay)

if jitter:
    jitter_range = delay * 0.25
    delay += random(-jitter_range, jitter_range)

is_retryable_status(status_code)

Check if HTTP status code is retryable.

Parameters:: status_code (int) – HTTP status code
Returns:: True if status code is retryable
Return type:: bool

is_max_retry(context)

Check if maximum retries reached.

Parameters:: context (RequestContext) – Request context
Returns:: True if max retries reached
Return type:: bool

Examples

from requestforge import ExponentialBackoffRetryStrategy

# Recommended production configuration
strategy = ExponentialBackoffRetryStrategy(
    max_retries=3,
    base_delay=1.0,
    max_delay=60.0,
    multiplier=2.0,
    jitter=True
)

# Custom retryable status codes
strategy = ExponentialBackoffRetryStrategy(
    max_retries=5,
    base_delay=1.0,
    retryable_status_codes=frozenset({
        408,  # Request Timeout
        429,  # Too Many Requests
        500,  # Internal Server Error
        502,  # Bad Gateway
        503,  # Service Unavailable
        504,  # Gateway Timeout
    })
)

# Aggressive retry (fast)
strategy = ExponentialBackoffRetryStrategy(
    max_retries=5,
    base_delay=0.5,
    max_delay=30.0,
    multiplier=2.0,
    jitter=True
)

# Conservative retry (slow)
strategy = ExponentialBackoffRetryStrategy(
    max_retries=3,
    base_delay=5.0,
    max_delay=300.0,
    multiplier=3.0,
    jitter=True
)

Retry Timeline Example:

Attempt 1: Immediate
Attempt 2: ~1s delay    (1.0 * 2^0 ± jitter)
Attempt 3: ~2s delay    (1.0 * 2^1 ± jitter)
Attempt 4: ~4s delay    (1.0 * 2^2 ± jitter)
Attempt 5: ~8s delay    (1.0 * 2^3 ± jitter)

CircuitBreakerRetryStrategy

class CircuitBreakerRetryStrategy(max_retries=3, failure_threshold=5, recovery_timeout=30.0, half_open_max_calls=3, base_delay=1.0)

Circuit breaker pattern implementation.

Parameters:

max_retries (int) – Retries per request
failure_threshold (int) – Failures before opening circuit
recovery_timeout (float) – Seconds before testing recovery
half_open_max_calls (int) – Test calls in half-open state
base_delay (float) – Base delay between retries

States

The circuit breaker has three states:

CLOSED: Normal operation, requests pass through
OPEN: Too many failures, reject requests immediately
HALF_OPEN: Testing if service recovered

Attributes

CircuitBreakerRetryStrategy.max_retries: int: Maximum retry attempts per request.

CircuitBreakerRetryStrategy.state: str: Current circuit state: ‘closed’, ‘open’, or ‘half_open’.

Methods

CircuitBreakerRetryStrategy.should_retry(context, exception)

Check if should retry considering circuit state.

Parameters:

context (RequestContext) – Request context
exception (Exception) – Exception that occurred

Returns:

True if should retry

Return type:

bool

CircuitBreakerRetryStrategy.get_delay(context)

Get delay based on attempt number.

Parameters:: context (RequestContext) – Request context
Returns:: Delay in seconds
Return type:: float

record_success(): Record a successful request (may close circuit).

reset(): Reset circuit breaker to CLOSED state.

Examples

from requestforge import CircuitBreakerRetryStrategy

strategy = CircuitBreakerRetryStrategy(
    max_retries=3,
    failure_threshold=5,     # Open after 5 failures
    recovery_timeout=30.0,   # Try recovery after 30s
    half_open_max_calls=3    # Test with 3 calls
)

config = builder.with_retry_strategy(strategy).build()
client = HttpClient(config)

# Check circuit state
print(strategy.state)  # 'closed', 'open', or 'half_open'

# Reset circuit manually
strategy.reset()

State Transitions:

CLOSED ──(5 failures)──> OPEN ──(30s timeout)──> HALF_OPEN
   ↑                                                  |
   └────────────(3 successful calls)─────────────────┘

Auth Retry Strategies

NoAuthRetryStrategy

class NoAuthRetryStrategy(auth_error_codes=None)

No authentication retry - fail immediately on auth error.

Parameters:: auth_error_codes (frozenset) – HTTP codes considered auth errors (default: {401})

SimpleAuthRetryStrategy

class SimpleAuthRetryStrategy(max_retries=1, delay=0.0, auth_error_codes=None)

Simple auth retry with fixed delay.

Parameters:

max_retries (int) – Maximum auth retry attempts
delay (float) – Delay between retries
auth_error_codes (frozenset) – Auth error codes (default: {401})

ExponentialAuthRetryStrategy

class ExponentialAuthRetryStrategy(max_retries=2, base_delay=0.5, max_delay=5.0, multiplier=2.0, jitter=True, auth_error_codes=None)

Exponential backoff for auth retries.

Parameters:

max_retries (int) – Maximum retry attempts
base_delay (float) – Initial delay
max_delay (float) – Maximum delay cap
multiplier (float) – Exponential multiplier
jitter (bool) – Add random jitter
auth_error_codes (frozenset) – Auth error codes

ConditionalAuthRetryStrategy

class ConditionalAuthRetryStrategy(max_retries=1, delay=0.0, should_retry_func=None, is_auth_error_func=None, auth_error_codes=None)

Conditional auth retry based on custom logic.

Parameters:

max_retries (int) – Maximum retry attempts
delay (float) – Delay between retries
should_retry_func (callable) – Custom function to determine if should retry
is_auth_error_func (callable) – Custom function to identify auth errors
auth_error_codes (frozenset) – Auth error codes

Example:

def should_retry(response):
    # Only retry if error is "token_expired"
    if response.status_code == 401:
        data = response.json_or_none()
        return data and data.get('error') == 'token_expired'
    return False

strategy = ConditionalAuthRetryStrategy(
    max_retries=2,
    should_retry_func=should_retry
)

Custom Retry Strategy

Implementing a custom retry strategy:

from requestforge.interfaces import RetryStrategyInterface
from requestforge.models import RequestContext

class CustomRetryStrategy(RetryStrategyInterface):
    """Custom retry strategy with business logic."""

    def __init__(self, max_retries=3):
        self._max_retries = max_retries

    @property
    def max_retries(self) -> int:
        return self._max_retries

    def should_retry(self, context: RequestContext, exception: Exception) -> bool:
        # Custom logic
        if context.attempt >= self._max_retries:
            return False

        # Only retry on specific days
        import datetime
        if datetime.datetime.now().weekday() >= 5:  # Weekend
            return False

        # Only retry timeout errors
        from requestforge import TimeoutException
        return isinstance(exception, TimeoutException)

    def get_delay(self, context: RequestContext) -> float:
        # Custom delay based on time of day
        import datetime
        hour = datetime.datetime.now().hour

        if 9 <= hour <= 17:  # Business hours
            return 1.0
        else:
            return 5.0  # Off-hours, longer delay

# Use custom strategy
config = builder.with_retry_strategy(CustomRetryStrategy()).build()

Retry Strategies API Reference

RetryStrategyInterface

Methods

Properties

NoRetryStrategy

SimpleRetryStrategy

Attributes

Methods

Examples

ExponentialBackoffRetryStrategy

Attributes

Methods

Examples

CircuitBreakerRetryStrategy

States

Attributes

Methods

Examples

Auth Retry Strategies

NoAuthRetryStrategy

SimpleAuthRetryStrategy

ExponentialAuthRetryStrategy

ConditionalAuthRetryStrategy

Custom Retry Strategy

See Also