Retry API Reference¶

def retry_if_exception(
    *exc_types: type[BaseException],
) -> Callable[[BaseException], bool]:
    """Build a predicate that matches specific exception types.

    Args:
        *exc_types: Exception classes to retry on.

    Returns:
        A callable ``(exc) -> bool``.
    """

    def predicate(exc: BaseException) -> bool:
        return isinstance(exc, exc_types)

    return predicate

def retry_if_result(predicate: Callable[[Any], bool]) -> Callable[[Any], bool]:
    """Mark a callable as a result-retry predicate (identity helper).

    Args:
        predicate: A callable ``(result) -> bool`` returning ``True`` to retry.

    Returns:
        The same callable, for self-documenting call sites.
    """
    return predicate

def retry_if_status(*status_codes: int) -> Callable[[BaseException], bool]:
    """Build a predicate that retries on HTTP status codes.

    Works with any exception carrying a ``status_code`` attribute
    (e.g. ``httpclient.HTTPError``).

    Args:
        *status_codes: HTTP status codes to retry on (e.g. 429, 502, 503).

    Returns:
        A callable ``(exc) -> bool``.
    """
    codes = set(status_codes)

    def predicate(exc: BaseException) -> bool:
        sc = getattr(exc, "status_code", None)
        return sc is not None and sc in codes

    return predicate

def retry(
    fn: Callable[..., Any] | None = None,
    *,
    max_retries: int = DEFAULT_MAX_RETRIES,
    base_delay: float = DEFAULT_BASE_DELAY,
    max_delay: float = DEFAULT_MAX_DELAY,
    backoff: str = "exponential",
    backoff_factor: float = DEFAULT_BACKOFF_FACTOR,
    jitter: str = "full",
    retry_on: tuple[type[BaseException], ...] | Callable[[BaseException], bool] = (
        Exception,
    ),
    retry_on_result: Callable[[Any], bool] | None = None,
    on_retry: Callable[[RetryState], None] | None = None,
) -> Callable[..., Any]:
    """Decorator that retries a function on failure with configurable backoff.

    Can be used with or without arguments::

        @retry
        def f(): ...

        @retry()
        def g(): ...

        @retry(max_retries=5, backoff="linear")
        def h(): ...

    Automatically detects async functions and uses ``asyncio.sleep``.

    Args:
        fn: The function to decorate (set automatically when used as
            ``@retry`` without parentheses).
        max_retries: Maximum number of retries (not counting the initial call).
        base_delay: Base delay in seconds before the first retry.
        max_delay: Upper bound on computed delay.
        backoff: Backoff strategy — ``"exponential"``, ``"linear"``, or
            ``"fixed"``.
        backoff_factor: Multiplier for exponential backoff.
        jitter: Jitter mode — ``"full"`` (uniform [0, delay]),
            ``"equal"`` (delay/2 + uniform [0, delay/2]), or ``"none"``.
        retry_on: Exception types or a callable ``(exc) -> bool`` deciding
            whether to retry.  Defaults to ``(Exception,)``.
        retry_on_result: Optional callable ``(result) -> bool``.  When it
            returns ``True`` the call is retried.
        on_retry: Optional callback invoked before each retry sleep with a
            :class:`RetryState` instance.

    Returns:
        The decorated function (sync or async, matching the original).

    Raises:
        RetryError: When retries are exhausted due to *retry_on_result*.
        The original exception: When retries are exhausted due to exceptions.
    """

    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        if inspect.iscoroutinefunction(fn):

            @functools.wraps(fn)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                return await _retry_async(
                    fn,
                    args,
                    kwargs,
                    max_retries=max_retries,
                    base_delay=base_delay,
                    max_delay=max_delay,
                    backoff=backoff,
                    backoff_factor=backoff_factor,
                    jitter=jitter,
                    retry_on=retry_on,
                    retry_on_result=retry_on_result,
                    on_retry=on_retry,
                )

            return async_wrapper

        @functools.wraps(fn)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            return _retry_sync(
                fn,
                args,
                kwargs,
                max_retries=max_retries,
                base_delay=base_delay,
                max_delay=max_delay,
                backoff=backoff,
                backoff_factor=backoff_factor,
                jitter=jitter,
                retry_on=retry_on,
                retry_on_result=retry_on_result,
                on_retry=on_retry,
            )

        return sync_wrapper

    if fn is not None:
        return decorator(fn)
    return decorator

def retry_call(
    fn: Callable[..., Any],
    args: tuple[Any, ...] = (),
    kwargs: dict[str, Any] | None = None,
    **retry_kwargs: Any,
) -> Any:
    """Call *fn* with retry logic without using a decorator.

    Args:
        fn: The callable to invoke.
        args: Positional arguments for *fn*.
        kwargs: Keyword arguments for *fn*.
        **retry_kwargs: Same keyword arguments accepted by :func:`retry`.

    Returns:
        The return value of *fn* on success.
    """
    if kwargs is None:
        kwargs = {}

    opts: dict[str, Any] = dict(
        max_retries=retry_kwargs.pop("max_retries", DEFAULT_MAX_RETRIES),
        base_delay=retry_kwargs.pop("base_delay", DEFAULT_BASE_DELAY),
        max_delay=retry_kwargs.pop("max_delay", DEFAULT_MAX_DELAY),
        backoff=retry_kwargs.pop("backoff", "exponential"),
        backoff_factor=retry_kwargs.pop("backoff_factor", DEFAULT_BACKOFF_FACTOR),
        jitter=retry_kwargs.pop("jitter", "full"),
        retry_on=retry_kwargs.pop("retry_on", (Exception,)),
        retry_on_result=retry_kwargs.pop("retry_on_result", None),
        on_retry=retry_kwargs.pop("on_retry", None),
    )

    if inspect.iscoroutinefunction(fn):
        return _retry_async(fn, args, kwargs, **opts)
    return _retry_sync(fn, args, kwargs, **opts)