Skip to content

yggdrasil.pyutils.retry

retry

retry.py

Lightweight, dependency-free retry decorators for sync and async Python code.

Features: - Configurable max attempts - Fixed or exponential backoff - Optional jitter - Optional total timeout - Works for both sync and async functions - Optional logging hook

retry

retry(
    exceptions: ExceptionTypes = Exception,
    tries: int = 3,
    delay: float = 0.5,
    backoff: float = 2.0,
    max_delay: float | None = None,
    jitter: Optional[Callable[[float], float]] = None,
    logger: Optional[Logger] = None,
    reraise: bool = True,
    timeout: float | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

Retry decorator that works for both sync and async functions.

Parameters

exceptions: Exception class or tuple of exception classes to catch and retry on. tries: Total number of attempts (including the first one). delay: Initial sleep delay between retries in seconds. backoff: Multiplier for exponential backoff (e.g. 2.0 doubles the delay each retry). max_delay: Upper bound for the delay. If None, delay is unbounded. jitter: Optional function taking the current delay and returning adjusted delay (e.g. lambda d: d * random.uniform(0.8, 1.2)). logger: Optional logger for debug/info messages. reraise: If True, reraise the last exception after exhausting retries or timeout. If False, returns None after all retries fail or timeout is hit. timeout: Optional max total time in seconds across all attempts. The timeout is checked after each failed attempt. In-flight attempts are not forcibly cancelled; we just stop scheduling new retries when the timeout is hit.

Example

@retry(exceptions=(ValueError,), tries=5, delay=0.1, backoff=2) def flaky(): ...

@retry(tries=5, timeout=2.0) async def async_flaky(): ...

retry_fixed

retry_fixed(
    exceptions: ExceptionTypes = Exception,
    tries: int = 3,
    delay: float = 0.5,
    **kwargs: Any
) -> Callable[[Callable[P, R]], Callable[P, R]]

Convenience wrapper: fixed delay, no backoff.

Example

@retry_fixed(tries=5, delay=1.0) def call_api(): ...

random_jitter

random_jitter(scale: float = 0.1) -> Callable[[float], float]

Returns a jitter function to add ± (scale * delay) randomness.

Example

@retry(jitter=random_jitter(0.2)) def flaky(): ...

flaky_function

flaky_function() -> str

Demonstrate retry behavior for a flaky sync function.

Returns:

Type Description
str

String result.

main async

main() -> None

Run an async retry demonstration.

Returns:

Type Description
None

None.