Retry

The Retry object.

AsyncRetry

Bases: BaseRetry[ReturnT]

Asynchronous retry

Examples:

>>> async def ok() -> bool:
...    return True
>>> retry = AsyncRetry[t.Awaitable[bool]](strategies=[StopAfterAttemptStrategy(20)])
>>> print(await retry(ok))

Source code in pyretries/retry.py

class AsyncRetry(BaseRetry[ReturnT]):
    """
    Asynchronous retry

    Examples:
        >>> async def ok() -> bool:
        ...    return True
        >>> retry = AsyncRetry[t.Awaitable[bool]](strategies=[StopAfterAttemptStrategy(20)])
        >>> print(await retry(ok))
    """

    async def _exec(self, state: RetryState[ReturnT]) -> None:
        """
        Executes `func` from `state`

        Args:
            state: Current retry state

        """
        assert inspect.iscoroutinefunction(
            state.func
        ), f"{self.__class__.__name__} needs an awaitable func"

        self._pre_exec(state)

        exception: Exception | None = None
        try:
            state.returned_value = await state.func(
                *(state.args or ()), **(state.kwargs or {})
            )
        except Exception as err:
            exception = err

        self._post_exec(state, exception)

    # NOTE: currently, the abtract method is a sync, and here we are defining async.
    # Pyright will alert about this, so we ignore the type for now until we
    # have a fix for it
    async def __call__(  # type:ignore
        self, func: FuncT[ReturnT], *args: t.Any, **kwargs: t.Any
    ) -> ReturnT | Exception | None:
        """
        Executes `func` and applies strategies

        Args:
            func: Address to function
            args: `func` non-positional arguments
            kwargs: `func` positional arguments

        Returns:
           Either `func`'s return value or None
        """
        state = RetryState[ReturnT](
            func=func,
            start_time=int(datetime.now().timestamp()),
            args=args,
            kwargs=kwargs,
        )

        if self.should_log:
            _logger.info(f"Calling '{func.__name__}'")

        should_reapply = True
        while should_reapply:
            await self._exec(state)

            if not (should_reapply := self._apply(state)):
                break

        self._save_state(state)
        return state.returned_value

__call__(func, *args, **kwargs) async

Executes func and applies strategies

Parameters:

Name	Type	Description	Default
func	FuncT[ReturnT]	Address to function	required
args	Any	func non-positional arguments	()
kwargs	Any	func positional arguments	{}

Returns:

Type	Description
ReturnT | Exception | None	Either func's return value or None

Source code in pyretries/retry.py

async def __call__(  # type:ignore
    self, func: FuncT[ReturnT], *args: t.Any, **kwargs: t.Any
) -> ReturnT | Exception | None:
    """
    Executes `func` and applies strategies

    Args:
        func: Address to function
        args: `func` non-positional arguments
        kwargs: `func` positional arguments

    Returns:
       Either `func`'s return value or None
    """
    state = RetryState[ReturnT](
        func=func,
        start_time=int(datetime.now().timestamp()),
        args=args,
        kwargs=kwargs,
    )

    if self.should_log:
        _logger.info(f"Calling '{func.__name__}'")

    should_reapply = True
    while should_reapply:
        await self._exec(state)

        if not (should_reapply := self._apply(state)):
            break

    self._save_state(state)
    return state.returned_value

BaseRetry

Bases: ABC, Generic[ReturnT]

Base class for all retry implementations. Requires __call__ implementation

Examples:

>>> Class RetryExample(BaseRetry[ReturnT]):
        ...
        def __call__(
            self,
            func: FuncT[ReturnT],
            *args: t.Tuple[t.Any],
            **kwargs: t.Dict[t.Any, t.Any],
        ) -> ReturnT | Exception | None:
             ...

Source code in pyretries/retry.py

class BaseRetry(abc.ABC, t.Generic[ReturnT]):
    """
    Base class for all retry implementations.
    Requires `__call__` implementation

    Examples:
        >>> Class RetryExample(BaseRetry[ReturnT]):
                ...
                def __call__(
                    self,
                    func: FuncT[ReturnT],
                    *args: t.Tuple[t.Any],
                    **kwargs: t.Dict[t.Any, t.Any],
                ) -> ReturnT | Exception | None:
                     ...

    """

    on_exceptions: set[type[Exception]] | None

    def __init__(
        self,
        strategies: t.Sequence[Strategy[ReturnT]] = [],
        on_exceptions: t.Sequence[type[Exception]] | None = None,
        before_hooks: t.Sequence[BeforeHookFuncT] | None = None,
        after_hooks: t.Sequence[AfterHookFuncT[ReturnT]] | None = None,
        retry_exception_hook: RetryExceptionCallHook | None = None,
        should_log: bool = True,
    ) -> None:
        """
        Args:
            strategies: Sequence of retry strategies
            on_exceptions: Sequence of exceptions to apply a retry strategy.
            before_hooks: Hooks to run before running `func`. Runs Before strategy.
            after_hooks: Hooks to run after running `func`. Runs before strategy.
            retry_exception_hook: Hook to run when `func` raised an exception. Runs before strategy.
            should_log: Specifies whether retry should log actions
        """
        self.strategies = list(reversed(strategies))
        self.on_exceptions = set(on_exceptions or []) or None

        self.before_hooks = before_hooks or []
        self.after_hooks = after_hooks or []
        self.retry_exception_hook = retry_exception_hook
        self.should_log = should_log

    @abc.abstractmethod
    def __call__(
        self,
        func: FuncT[ReturnT],
        *args: t.Tuple[t.Any],
        **kwargs: t.Dict[t.Any, t.Any],
    ) -> ReturnT | Exception | None:
        """
        Executes `func` and applies strategies

        Args:
            func: Address to function
            args: `func` non-positional arguments
            kwargs: `func` positional arguments

        """
        raise NotImplementedError

    def _save_state(self, state: RetryState[ReturnT]) -> None:
        """
        Saves retry state to `func`

        Args:
            state: Current retry state
        """
        setattr(state.func, "retry_state", state)

    def _exec_strategy(self, state: RetryState[ReturnT]):
        """
        Applies user defined strategies.

        Args:
            state: Current retry state

        Raises:
            RetryExaustedError: Raised when strategy is exausted of no strategy is available
        """
        if state.strategy is None:
            if len(self.strategies):
                state.strategy = self.strategies.pop()
            else:
                raise RetryExaustedError

        try:
            if state.strategy.should_stop:
                raise RetryStrategyExausted

            if self.should_log:
                _logger.info(
                    f"Executing '{state.strategy.__class__.__name__}' retry strategy. "
                    f"Current attempt {state.current_attempts}"
                )

            state.strategy.eval(state.returned_value)
            state.current_attempts += 1

            if state.strategy.should_stop:
                state.strategy = None

        except RetryStrategyExausted:
            raise RetryExaustedError

    def _apply(self, state: RetryState[ReturnT]) -> bool:
        """
        Checks if last state raised an exception and executes the next available strategy

        Raise `RetryExaustedError` if last executes strategy raised

        Args:
            state: Current retry state

        Raises:
            RetryExaustedError: Raised when `func` exception is not defined in `on_exceptions` sequence
                                or when strategy raises
        """
        try:
            if not state.raised:
                return False

            if (exc := state.exception.__class__) not in (self.on_exceptions or [exc]):
                raise RetryExaustedError from state.exception

            self._exec_strategy(state)
            state.clear()
            return True

        except RetryExaustedError as err:
            raise err from state.exception if state.raised else None

    def _pre_exec(self, _: RetryState[ReturnT]) -> None:
        """
        Should be called before running `func`

        Args:
            state: Current retry state. Currently not used.
        """
        for hook in self.before_hooks:
            hook()

    def _post_exec(
        self, state: RetryState[ReturnT], exception: Exception | None
    ) -> None:
        """
        Should be called after running `func`.
        If exception was raised, this should be passed in `exception` argument.

        Args:
            state: Current retry state. Currently not used.
            exception: Raised exception
        """
        if exception:
            state.exception = exception

            if self.retry_exception_hook:
                self.retry_exception_hook(exception)

        for hook in self.after_hooks:
            hook(state.exception or state.returned_value)

        state.end_time = int(datetime.now().timestamp())

__call__(func, *args, **kwargs) abstractmethod

Executes func and applies strategies

Parameters:

Name	Type	Description	Default
func	FuncT[ReturnT]	Address to function	required
args	Tuple[Any]	func non-positional arguments	()
kwargs	Dict[Any, Any]	func positional arguments	{}

Source code in pyretries/retry.py

@abc.abstractmethod
def __call__(
    self,
    func: FuncT[ReturnT],
    *args: t.Tuple[t.Any],
    **kwargs: t.Dict[t.Any, t.Any],
) -> ReturnT | Exception | None:
    """
    Executes `func` and applies strategies

    Args:
        func: Address to function
        args: `func` non-positional arguments
        kwargs: `func` positional arguments

    """
    raise NotImplementedError

__init__(strategies=[], on_exceptions=None, before_hooks=None, after_hooks=None, retry_exception_hook=None, should_log=True)

Parameters:

Name	Type	Description	Default
strategies	Sequence[Strategy[ReturnT]]	Sequence of retry strategies	[]
on_exceptions	Sequence[type[Exception]] | None	Sequence of exceptions to apply a retry strategy.	None
before_hooks	Sequence[BeforeHookFuncT] | None	Hooks to run before running func. Runs Before strategy.	None
after_hooks	Sequence[AfterHookFuncT[ReturnT]] | None	Hooks to run after running func. Runs before strategy.	None
retry_exception_hook	RetryExceptionCallHook | None	Hook to run when func raised an exception. Runs before strategy.	None
should_log	bool	Specifies whether retry should log actions	True

Source code in pyretries/retry.py

def __init__(
    self,
    strategies: t.Sequence[Strategy[ReturnT]] = [],
    on_exceptions: t.Sequence[type[Exception]] | None = None,
    before_hooks: t.Sequence[BeforeHookFuncT] | None = None,
    after_hooks: t.Sequence[AfterHookFuncT[ReturnT]] | None = None,
    retry_exception_hook: RetryExceptionCallHook | None = None,
    should_log: bool = True,
) -> None:
    """
    Args:
        strategies: Sequence of retry strategies
        on_exceptions: Sequence of exceptions to apply a retry strategy.
        before_hooks: Hooks to run before running `func`. Runs Before strategy.
        after_hooks: Hooks to run after running `func`. Runs before strategy.
        retry_exception_hook: Hook to run when `func` raised an exception. Runs before strategy.
        should_log: Specifies whether retry should log actions
    """
    self.strategies = list(reversed(strategies))
    self.on_exceptions = set(on_exceptions or []) or None

    self.before_hooks = before_hooks or []
    self.after_hooks = after_hooks or []
    self.retry_exception_hook = retry_exception_hook
    self.should_log = should_log

Retry

Bases: BaseRetry[ReturnT]

Synchronous retry

Examples:

>>> def ok() -> bool:
...    return True
>>> retry = Retry[bool](strategies=[StopAfterAttemptStrategy(20)])
>>> print(retry(ok))

Source code in pyretries/retry.py

class Retry(BaseRetry[ReturnT]):
    """
    Synchronous retry

    Examples:
        >>> def ok() -> bool:
        ...    return True
        >>> retry = Retry[bool](strategies=[StopAfterAttemptStrategy(20)])
        >>> print(retry(ok))
    """

    def exec(self, state: RetryState[ReturnT]) -> None:
        """
        Executes `func` from `state`

        Args:
            state: Current retry state

        """
        self._pre_exec(state)

        exception: Exception | None = None
        try:
            state.returned_value = state.func(
                *(state.args or ()), **(state.kwargs or {})
            )
        except Exception as err:
            exception = err

        self._post_exec(state, exception)

    def __call__(
        self, func: FuncT[ReturnT], *args: t.Any, **kwargs: t.Any
    ) -> ReturnT | Exception | None:
        """
        Executes `func` and applies strategies

        Args:
            func: Address to function
            args: `func` non-positional arguments
            kwargs: `func` positional arguments

        Returns:
           Either `func`'s return value or None
        """
        state = RetryState[ReturnT](
            func=func,
            start_time=int(datetime.now().timestamp()),
            args=args,
            kwargs=kwargs,
        )

        if self.should_log:
            _logger.info(f"Calling '{func.__name__}'")

        should_reapply = True
        while should_reapply:
            self.exec(state)

            if not (should_reapply := self._apply(state)):
                break

        self._save_state(state)
        return state.returned_value

__call__(func, *args, **kwargs)

Executes func and applies strategies

Parameters:

Name	Type	Description	Default
func	FuncT[ReturnT]	Address to function	required
args	Any	func non-positional arguments	()
kwargs	Any	func positional arguments	{}

Returns:

Type	Description
ReturnT | Exception | None	Either func's return value or None

Source code in pyretries/retry.py

def __call__(
    self, func: FuncT[ReturnT], *args: t.Any, **kwargs: t.Any
) -> ReturnT | Exception | None:
    """
    Executes `func` and applies strategies

    Args:
        func: Address to function
        args: `func` non-positional arguments
        kwargs: `func` positional arguments

    Returns:
       Either `func`'s return value or None
    """
    state = RetryState[ReturnT](
        func=func,
        start_time=int(datetime.now().timestamp()),
        args=args,
        kwargs=kwargs,
    )

    if self.should_log:
        _logger.info(f"Calling '{func.__name__}'")

    should_reapply = True
    while should_reapply:
        self.exec(state)

        if not (should_reapply := self._apply(state)):
            break

    self._save_state(state)
    return state.returned_value

exec(state)

Executes func from state

Parameters:

Name	Type	Description	Default
state	RetryState[ReturnT]	Current retry state	required

Source code in pyretries/retry.py

def exec(self, state: RetryState[ReturnT]) -> None:
    """
    Executes `func` from `state`

    Args:
        state: Current retry state

    """
    self._pre_exec(state)

    exception: Exception | None = None
    try:
        state.returned_value = state.func(
            *(state.args or ()), **(state.kwargs or {})
        )
    except Exception as err:
        exception = err

    self._post_exec(state, exception)

RetryState dataclass

Bases: Generic[ReturnT]

Stores current retry state

Parameters:

Name	Type	Description	Default
func	FuncT[ReturnT]	Function address to retry	required
args	Sequence[Any] | None	func non-positional arguments	None
kwargs	Dict[str, Any] | None	func positional arguments	None
start_time	int	Timestamp when retry first started	required
end_time	int	Timestamp when retry ended	0
strategy_func		Next strategy to run	required
current_attempts	int	Number of current retry attempts	0
exception	Exception | None	Exception raised by func	None

Source code in pyretries/retry.py

@dataclass
class RetryState(t.Generic[ReturnT]):
    """Stores current retry state

    Args:
        func: Function address to retry
        args: `func` non-positional arguments
        kwargs: `func` positional arguments
        start_time: Timestamp when retry first started
        end_time: Timestamp when retry ended
        strategy_func: Next strategy to run
        current_attempts: Number of current retry attempts
        exception: Exception raised by `func`

    """

    func: FuncT[ReturnT]
    start_time: int
    end_time: int = 0
    strategy: Strategy[ReturnT] | None = None
    args: t.Sequence[t.Any] | None = None
    kwargs: t.Dict[str, t.Any] | None = None
    current_attempts: int = 0
    returned_value: ReturnT | None = None
    exception: Exception | None = None

    @property
    def raised(self) -> bool:
        """Checks if `func` raised an exception"""
        return isinstance(self.exception, Exception)

    def clear(self) -> None:
        """Clears retry state"""
        self.exception = None
        self.returned_value = None

    def __repr__(self) -> str:
        cls_name = type(self).__name__
        return (
            f"{cls_name}(start_time={self.start_time}, "
            f"end_time={self.end_time}, "
            f"current_attempts={self.current_attempts}, "
            f"exception={repr(self.exception)}), "
            f"returned_value={self.returned_value})"
        )

raised: bool property

Checks if func raised an exception

clear()

Clears retry state

Source code in pyretries/retry.py

def clear(self) -> None:
    """Clears retry state"""
    self.exception = None
    self.returned_value = None

retry(strategies=[], on_exceptions=None, before_hooks=None, after_hooks=None, retry_exception_hook=None, should_log=True)

Retry decorator. Works both for sync and async functions

Examples:

>>> @retry(strategies=[strategy.NoopStrategy(1)])
... def ok() -> bool:
...     return True
>>> ok()
INFO:retries.retry:Calling 'ok'
True

Parameters:

Name	Type	Description	Default
strategies	Sequence[Strategy]	Sequence of retry strategies	[]
on_exceptions	Sequence[type[Exception]] | None	Sequence of exceptions to apply a retry strategy.	None
before_hooks	Sequence[BeforeHookFuncT] | None	Hooks to run before running func. Runs Before strategy.	None
after_hooks	Sequence[AfterHookFuncT[ReturnT]] | None	Hooks to run after running func. Runs before strategy.	None
retry_exception_hook	RetryExceptionCallHook | None	Hook to run when func raised an exception. Runs before strategy.	None
should_log	bool	Specifies whether retry should log actions	True

Returns:

Name	Type	Description
func	FuncT[ReturnT]	Functions return value or exception

Source code in pyretries/retry.py

def retry(
    strategies: t.Sequence[Strategy] = [],
    on_exceptions: t.Sequence[type[Exception]] | None = None,
    before_hooks: t.Sequence[BeforeHookFuncT] | None = None,
    after_hooks: t.Sequence[AfterHookFuncT[ReturnT]] | None = None,
    retry_exception_hook: RetryExceptionCallHook | None = None,
    should_log: bool = True,
):
    """
    Retry decorator. Works both for sync and async functions

    Examples:
        >>> @retry(strategies=[strategy.NoopStrategy(1)])
        ... def ok() -> bool:
        ...     return True
        >>> ok()
        INFO:retries.retry:Calling 'ok'
        True

    Args:
        strategies: Sequence of retry strategies
        on_exceptions: Sequence of exceptions to apply a retry strategy.
        before_hooks: Hooks to run before running `func`. Runs Before strategy.
        after_hooks: Hooks to run after running `func`. Runs before strategy.
        retry_exception_hook: Hook to run when `func` raised an exception. Runs before strategy.
        should_log: Specifies whether retry should log actions

    Returns:
        func (FuncT[ReturnT]): Functions return value or exception
    """

    def decorator_retry(
        func: FuncT[t.Any],
    ) -> FuncT[ReturnT]:
        @functools.wraps(func)
        def wrapper_retry(*args, **kwargs):
            if inspect.iscoroutinefunction(func):
                return AsyncRetry(
                    strategies=strategies,
                    on_exceptions=on_exceptions,
                    before_hooks=before_hooks,
                    after_hooks=after_hooks,
                    retry_exception_hook=retry_exception_hook,
                    should_log=should_log,
                )(func, *args, **kwargs)

            return Retry(
                strategies=strategies,
                on_exceptions=on_exceptions,
                before_hooks=before_hooks,
                after_hooks=after_hooks,
                retry_exception_hook=retry_exception_hook,
                should_log=should_log,
            )(func, *args, **kwargs)

        return t.cast(FuncT[ReturnT], wrapper_retry)

    return decorator_retry