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.