Module

utils.retry

Retry utilities with exponential backoff.

Provides both synchronous and asynchronous retry decorators and functions with configurable backoff strategies and jitter.

Example:

from bengal.utils.retry import retry_with_backoff, calculate_backoff

# Retry function with backoff
result = retry_with_backoff(
    fetch_data,
    retries=3,
    base_delay=0.5,
    exceptions=(ConnectionError, TimeoutError),
)

# Calculate backoff for custom retry loop
delay = calculate_backoff(attempt=2, base=0.5, max_delay=10.0)

Related Modules:

  • bengal.health.linkcheck.async_checker: Uses for HTTP retry
  • bengal.utils.file_lock: Uses for lock acquisition retry
View source
3 Functions

Functions

calculate_backoff
Calculate exponential backoff delay with optional jitter. Uses formula: base * (2 ^ attempt) with …
4 float 
def calculate_backoff(attempt: int, base: float = 0.5, max_delay: float = 10.0, jitter: bool = True) -> float

Calculate exponential backoff delay with optional jitter.

Uses formula: base * (2 ^ attempt) with ±25% jitter.

Parameters 4

Name Type Default Description
attempt int —

Current attempt number (0-indexed)
base float 0.5

Base delay in seconds
max_delay float 10.0

Maximum delay cap
jitter bool True

Add random jitter to prevent thundering herd

Returns

float —

Delay in seconds

retry_with_backoff
Execute function with retry and exponential backoff.
7 T 
def retry_with_backoff(func: Callable[[], T], retries: int = 3, base_delay: float = 0.5, max_delay: float = 10.0, jitter: bool = True, exceptions: tuple[type[Exception], ...] = (Exception,), on_retry: Callable[[int, Exception], None] | None = None) -> T

Execute function with retry and exponential backoff.

Parameters 7

Name Type Default Description
func Callable[[], T] —

Function to execute (no arguments)
retries int 3

Maximum retry attempts
base_delay float 0.5

Base delay between retries
max_delay float 10.0

Maximum delay cap
jitter bool True

Add jitter to prevent thundering herd
exceptions tuple[type[Exception], ...] (Exception,)

Exception types to catch and retry
on_retry Callable[[int, Exception], None] | None None

Optional callback(attempt, exception) on each retry

Returns

T —

Result of successful function call

async_retry_with_backoff async
Execute async function with retry and exponential backoff.
7 T 
async def async_retry_with_backoff(coro_func: Callable[[], Awaitable[T]], retries: int = 3, base_delay: float = 0.5, max_delay: float = 10.0, jitter: bool = True, exceptions: tuple[type[Exception], ...] = (Exception,), on_retry: Callable[[int, Exception], None] | None = None) -> T

Execute async function with retry and exponential backoff.

Parameters 7

Name Type Default Description
coro_func Callable[[], Awaitable[T]] —

Async function to execute (no arguments, returns awaitable)
retries int 3

Maximum retry attempts
base_delay float 0.5

Base delay between retries
max_delay float 10.0

Maximum delay cap
jitter bool True

Add jitter to prevent thundering herd
exceptions tuple[type[Exception], ...] (Exception,)

Exception types to catch and retry
on_retry Callable[[int, Exception], None] | None None

Optional callback(attempt, exception) on each retry

Returns

T —

Result of successful coroutine