Tenacity

Tenacity#

Tenacity is a package that allows to wrap a Python callable with uncertain output. It provies built-in features, so you don’t need to implement them by yourself, while still offerting high degree of flexibility.

import tenacity

Retrying#

The tenacity.Retrying object is a centract object in the package API. It retains the retry settings and executes a given callable according to the defined retry policies.

API of this object allows to specify:

  • When to stop retrying.

  • How much to wait.

  • Conditions to retry.

The popular way to define a callable that requires a retry is tenacity.retry, which simpy wraps a given callable in Retry object.

Consider the usage of the Retyring code, which defines Retrying and specifies the ways in which it stops attempts and waits for the next one.

exmaple_retrying = tenacity.Retrying(
    stop=tenacity.stop_after_delay(3),
    wait=tenacity.wait_random(1, 5)
)

Consider the function that outputs the time it was called:

from datetime import datetime


def some_function():
    print("Executing")
    print(datetime.now().strftime("%H:%M:%S"))
    raise ValueError

As Retrying is set to stop retrying after 3 seconds and to make random stops between attempts, the number of retries and the period between them could be different:

try:
    exmaple_retrying(some_function)
except Exception as e:
    print(e)

Executing
00:06:39
Executing
00:06:42
Executing
00:06:46
RetryError[<Future at 0x71ec134f5f90 state=finished raised ValueError>]

try:
    exmaple_retrying(some_function)
except Exception as e:
    print(e)

Executing
00:06:48
Executing
00:06:52
RetryError[<Future at 0x71ec134563f0 state=finished raised ValueError>]

Whether to retry#

The retry argument of Retrying allows to specify the conditions under which the object would initiate a new try to execute the given callable.

The prebuild ones:

  • tenacity.retry_if_exception_message.

  • tenacity.retry_if_exception_type.

  • tenacity.retry_if_not_exception_message.

  • tenacity.retry_if_no_exception_type.

Use the tenacity.retry_if_exception function to define a custom function that descides whether a retry is needed. The function will be called with case of exception, and the excepiton object would be passed as an argument. The output of True/False defines whether the framework will produce a retry.

Check the:

The following cell shows the actual outputs of prebuilt functions:

type_of_retry = type(tenacity.retry_if_exception_type(ValueError))
print(type_of_retry, end='\n\n')

for v in type_of_retry.__mro__:
    print(v)

<class 'tenacity.retry.retry_if_exception_type'>

<class 'tenacity.retry.retry_if_exception_type'>
<class 'tenacity.retry.retry_if_exception'>
<class 'tenacity.retry.retry_base'>
<class 'abc.ABC'>
<class 'object'>

The following cell provides an example of a custom function definition, showing the type of output that the tenacity function passes to the decision function:

def if_retry(inp):
    print("Input")
    print(type(inp))
    print(inp)
    return False


@tenacity.retry(
    retry=tenacity.retry_if_exception(if_retry)
)
def some_function(is_raise):
    if is_raise:
        raise ValueError("The raised exception")
    else:
        return

The exception raised case:

try:
    some_function(True)
except:
    pass

Input
<class 'ValueError'>
The raised exception

If an exception is not raised, the if_retry function is not even called:

some_function(False)

Waiting#

The wait parameter allows you to define the framework’s waiting strategy before attempting.

Check the Wait Funcitons to learn about API and predefined strategies.

The following cell defines the some_function with custom wait strategy. The output of the wait_strategy determines the time between attempts, in seconds:

from datetime import datetime


def wait_strategy(wait):
    print("Waiting strategy defition")
    print(type(wait))
    print(wait)
    return 2


@tenacity.retry(
    wait=wait_strategy,
    stop=tenacity.stop_after_attempt(2)
)
def some_function():
    print(datetime.now().strftime('%M.%S'))
    raise ValueError

The calls of the wait_strategy shows the type and content of the input argument.

try:
    some_function()
except:
    pass

06.18
Waiting strategy defition
<class 'tenacity.RetryCallState'>
<RetryCallState 128122022891920: attempt #1; slept for 0.0; last result: failed (ValueError )>
06.20
Waiting strategy defition
<class 'tenacity.RetryCallState'>
<RetryCallState 128122022891920: attempt #2; slept for 2.0; last result: failed (ValueError )>

Note that the interval between calls to the some_function is exactly 2 seconds, as specified by the wait_strategy.

Contents