Retry Module¶
forge.retry — Exponential backoff, jitter, and circuit breaker for resilient
external service calls.
Overview¶
The Retry module handles transient failures in external service calls with configurable backoff strategies, circuit breaker protection, and automatic logging of retry attempts. Works on async functions and as an async context manager.
Installation¶
Quick Start¶
from forge.retry import retry
@retry(attempts=3, backoff="exponential")
async def fetch_data(url: str):
# If this raises an exception, it will be retried
# with exponential backoff (plus jitter)
return await http_client.get(url)
Key Features¶
Backoff Strategies¶
from forge.retry import retry
# Exponential backoff with full jitter (default)
@retry(attempts=5, backoff="exponential")
# Linear backoff
@retry(attempts=5, backoff="linear")
# Constant delay
@retry(attempts=5, backoff="constant")
Circuit Breaker¶
from forge.retry import CircuitBreaker
breaker = CircuitBreaker(
failure_threshold=5, # Open after 5 failures
recovery_time=60.0, # Try again after 60 seconds
)
@retry(attempts=3, circuit_breaker=breaker)
async def call_external_api():
...
Manual Retry Context¶
from forge.retry import attempt, CircuitBreaker
breaker = CircuitBreaker(failure_threshold=5, recovery_time=60)
async with attempt(max_attempts=3, circuit_breaker=breaker) as ctx:
result = await call_external_api()
if not result.is_success:
ctx.retry() # Trigger a retry
Exception Filtering¶
@retry(
attempts=3,
exceptions=[ConnectionError, TimeoutError],
)
async def call_api():
# Only retry on ConnectionError and TimeoutError
# Other exceptions pass through immediately
...
API Reference¶
Retry and resilience module — exponential backoff, jitter, and circuit breaker.
Provides the @retry decorator and CircuitBreaker class for handling
transient failures in external service calls. Supports configurable backoff
strategies, exception filtering, and automatic logging.
Classes¶
CircuitBreaker ¶
CircuitBreakerOpenError ¶
Raised when a call is rejected because the circuit breaker is open.
NonRetryableError ¶
Raise this inside a retried call to abort retrying immediately.
RetryError ¶
Raised when all retry attempts have been exhausted.
Attributes¶
original_exception : The last exception raised by the wrapped callable. attempt_count : Total number of attempts made (including the initial call). total_delay : Total wall-clock time spent sleeping between retries.
RetryModule ¶
Methods:¶
circuit_breaker ¶
circuit_breaker(failure_threshold: int | None = None, recovery_time: float | None = None, *, name: str = '') -> Any
Return a :class:~forge.retry.circuit.CircuitBreaker with module defaults.
retry ¶
retry(fn: Callable[..., Any] | None = None, *, attempts: int | None = None, backoff: str | None = None, base_delay: float | None = None, max_delay: float | None = None, jitter: bool | None = None, timeout: float | None = None, retryable_exceptions: tuple[type[Exception], ...] | None = None) -> Any
Return a :func:retry pre-configured with module defaults.
Functions:¶
constant ¶
Constant backoff — always returns base_delay.
exponential ¶
Exponential backoff with full jitter.
Delay = U(0, min(base_delay * 2 ** (attempt - 1), max_delay))
get_backoff ¶
Return the backoff function for the given strategy name.
Raises ValueError for unknown strategies.
linear ¶
Linear backoff.
Delay = min(base_delay * attempt, max_delay)
retry ¶
retry(fn: Callable[..., Any] | None = None, *, attempts: int = 3, backoff: str = 'exponential', base_delay: float = 1.0, max_delay: float = 60.0, jitter: bool = True, timeout: float | None = None, retryable_exceptions: tuple[type[Exception], ...] | None = None) -> Any
Async retry — use as a decorator or async context manager.
Decorator form::
@retry(attempts=3)
async def fetch() -> bytes:
...
Context manager form::
async with retry(attempts=3):
data = await fetch()