This article gives a brief overview of the Tenacity library, and how it helps to retry actions in case of transient failures like loss of network.
All code should be able to handle transient failures to connect to a service, loss of network, timeouts, etc, gracefully, by providing a retry mechanism that remediates the condition and provides the details of these failures. This remediate mechanism could be as simple as providing a delay before retrying the same action.
Tenacity simplifies the action of retrying behaviour. It offers a decorator based approach, which provides retry behaviour for any condition.
We can easily add the package with a pip installation.
pip install tenacity
We may need to perform a retry in our code for various reasons. Some might be as simple as looping for a duration or to perform a retry only for a specific error type. Let us discuss the various types of retry approaches provided by the Tenacity library.
Basic retry
A simple retry will help to have a retry whenever an exception is raised. The function retries until there is no exception in the code flow. This approach may not be perfect, but a great one to start experimenting with in the library.
@retry
def simple_retry_forever():
print(“simple retry which runs forever”)
raise Exception(“Simple retry forever”)
Retry with time and number of attempts
When developers start working out retries, they typically set a time period or limit the number of retry attempts as part of a loop.
Retry with number of attempts: Here the code tries to retry for a maximum of five attempts before throwing a retry error:
import tenacity
from tenacity import *
@retry(stop=stop_after_delay(20))
def retry_stop_after_20_s():
print(“Stopping after 20 seconds”)
raise Exception(“retry until 20 sec is completed”)
Waiting before retries
Many times, we may need to wait before the retry is attempted. This feature is specifically useful in areas where throttling of requests needs to be handled.
import tenacity
from tenacity import *
@retry(wait=wait_fixed(2),stop=(stop_after_delay(12) | stop_after_attempt(6)))
def retry_wait_between_5_s():
print(“Wait 2 seconds between retries”)
raise Exception(“retry”)
In the above example, we have combined wait and attempt levels along with the duration of retry. In this case, the code will try to retry six times, post which a retry error will be raised.
There are different strategies of waits that are adopted like wait_random, wait_exponential, wait_chain, etc. You can explore these and decide the best choice for your application code.
Condition for retry
Many times, we would like the retry to happen only for a certain condition or error message. For example, we may want to retry a function only if we have an IOError raised.
import tenacity
from tenacity import *
import random
@retry(retry=retry_if_exception_type(IOError))
def retry_io_error():
print(“Retry forever with no wait if an IOError occurs, raise any other errors”)
raise random.choice([IOError,Exception])
retry_io_error()
The output is shown in Figure 1.
In the above example, the retry is attempted only if IOError is raised. Likewise, there are multiple other conditional retry strategies like retry_if_not_exception_type, retry_if_exception_message, retry_if_not_exception_type, etc.
Changing retry behaviour based on function outcome
We can extend the retry condition based on the outcome of another function. This approach is of immense value, as we can try to implement a remediation strategy for the application for the retry attempt to be made.
import requests
import random
def validate_live():
choice_result = [True,False]
print(“bring the choice of result”)
return random.choice(choice_result)
@retry(retry=retry_if_result(validate_live))
def execute_return_none():
print(“Retry with no wait if the return value is True”)
execute_return_none()
print(“Retry with no wait if the return value is True”)
execute_return_none()
Note that we haven’t raised any exception in the above. However, the retry loop will be triggered based on the outcome of the function. In this example, it is validate_live.
The output will be:
/usr/local/bin/python3 /Users/Srikanth_Mohan/learning/learn_tenacity.py Retry with no wait if return value is True bring the choice of result Retry with no wait if return value is True bring the choice of result Retry with no wait if return value is True bring the choice of result Process finished with exit code 0
Retrying for code block
Many times, we would like the retry to be implemented within a code block rather than wrapping it at the function level.
from tenacity import Retrying, RetryError, stop_after_attempt
try:
for attempt in Retrying(stop=stop_after_attempt(3)):
with attempt:
raise Exception(‘My code is failing!’)
except RetryError:
pass
Providing details on the retries
We can get the details on the retries made over a function by using the retry attribute attached to the
import tenacity
from tenacity import *
import random
@retry(retry=retry_if_exception_type(IOError),stop=stop_after_attempt(3))
def retry_io_error():
print(“Retry forever with no wait if an IOError occurs, raise any other errors”)
raise random.choice([IOError,Exception])
try:
retry_io_error()
except Exception:
pass
print (retry_io_error.retry.statistics)
The output will be as shown below:
/usr/local/bin/python3 /Users/Srikanth_Mohan/learning/learn_tenacity.py
Retry forever with no wait if an IOError occurs, raise any other errors
Retry forever with no wait if an IOError occurs, raise any other errors
{‘start_time’: 330092.244061281, ‘attempt_number’: 2, ‘idle_for’: 0, ‘delay_since_first_attempt’: 3.829202614724636e-05}
Process finished with exit code 0
We have covered some of the important features required for the retry mechanism using Tenacity. This library provides further functionalities like handling the retry failures using custom callbacks, adding logging before and after the retries, etc. It provides support retries for async services as well. I do hope this learning will come in handy in your work.
