Retry Functions

The retry functions can be used to do automatic retries. There are three different ways to use them. htf.retry can be called directly. htf.retriable is a decorator that can be used to decorate functions so that they are retried and htf.retrying is a context manager that can be used to do retries within a context block.

If an exception occurs during execution of the action, it is retried until the maximum number of attempts is reached. Exceptions during the last retry are re-raised to inform the caller of its occurence.

The delay between attempts can also be specified. By default there is no delay.

Direct Calls

Direct calls are done using htf.retry or htf.retry_sync.

Using coroutines you can run:

import htf

count = 0

async def foo(bar):
    global count

    print("foo() called with", bar)
    print("count:", count)

    count += 1

    if count < 3:
        raise ValueError("count is too small")

    return "success"

await htf.retry(foo, args=("bar",), attempts=3, delay=1.0)

To avoid using coroutines, you can run:

import htf

count = 0

def foo(bar):
    global count

    print("foo() called with", bar)
    print("count:", count)

    count += 1

    if count < 3:
        raise ValueError("count is too small")

    return "success"

htf.retry_sync(foo, args=("bar",), attempts=3, delay=1.0)

async htf.retry(action: ~typing.Callable[[...], ~typing.Any], args: ~typing.Tuple | None = None, kwargs: ~typing.Dict[str, ~typing.Any] | None = None, attempts: int = 2, delay: float | None = None, retry_exceptions: ~typing.Tuple[~typing.Any] = (<class 'BaseException'>,)) → Any

Retry action with a given number of attempts and a given delay.

Parameters:

action – the action to be retried (coroutines are supported)
args=() – the positional arguments passed to action
kwargs={} – the keyword arguments passed to action
attempts=2 – the number of attempts
delay=None – the delay between attempts in seconds
retry_exceptions= (BaseException,) – the exceptions to be catched to retry the action

Returns:

the return value of action(*args, **kwargs)

htf.retry_sync(action: ~typing.Callable[[...], ~typing.Any], args: ~typing.Tuple | None = None, kwargs: ~typing.Dict[str, ~typing.Any] | None = None, attempts: int = 2, delay: float | None = None, retry_exceptions: ~typing.Tuple[~typing.Any] = (<class 'BaseException'>,)) → Any

Retry action with a given number of attempts and a given delay.

Parameters:

action – the action to be retried (coroutines are not supported)
args=() – the positional arguments passed to action
kwargs={} – the keyword arguments passed to action
attempts=2 – the number of attempts
delay=None – the delay between attempts in seconds
retry_exceptions= (BaseException,) – the exceptions to be catched to retry the action

Returns:

the return value of action(*args, **kwargs)

Using the Decorator

If a function or method is supposed to always do retries, the decorator htf.retriable can be used. If the decorated method is a coroutine the result is a coroutine, too.

import htf

count = 0

@htf.retriable(attempts=3, delay=1.0)
def foo(bar):
    global count

    print("foo() called with", bar)
    print("count:", count)

    count += 1

    if count < 3:
        raise ValueError("count is too small")

    return "success"


foo("bar")

htf.retriable(attempts: int = 2, delay: float | None = None, retry_exceptions: ~typing.Tuple[~typing.Any] = (<class 'BaseException'>, )) → Callable[[Callable[[...], Any]], Any]

A decorator factory for retry(). Wrap your function in @retriable(…) to give it retry powers!

Parameters:

attempts=2 – the number of attempts
delay=None – the delay between attempts in seconds
retry_exceptions= (BaseException,) – the exceptions to be catched to retry the action

Returns:

a function decorator supporting retries. If a coroutine is supplied a coroutine is returned: else not.

Using the Context Manager

To use retries within a context block use the htf.retrying context manager.

import htf

count = 0

def foo(bar):
    global count

    print("foo() called with", bar)
    print("count:", count)

    count += 1

    if count < 3:
        raise ValueError("count is too small")

    return "success"


with htf.retrying(foo, attempts=3, delay=1.0) as retrying_foo:
    await retrying_foo("bar")

htf.retrying(action: ~typing.Callable[[...], ~typing.Any], attempts: int = 2, delay: float | None = None, retry_exceptions: ~typing.Tuple[~typing.Any] = (<class 'BaseException'>,)) → ContextManager | AsyncContextManager

A context manager for wrapping action with retry functionality with a given number of attempts and a given delay.

Parameters:

action – the action to be retried
attempts=2 – the number of attempts
delay=None – the delay between attempts in seconds
retry_exceptions= (BaseException,) – the exceptions to be catched to retry the action

Returns:

a context manager supporting retries. If a coroutine is supplied an async context manager: is returned.