Sometimes, a operate is named with dangerous inputs or in a foul program state, so it fails. In languages like Python, this normally leads to an exception.
But generally exceptions are brought on by completely different points or are transitory. Imagine code that should hold working within the face of caching knowledge being cleaned up. In idea, the code and the cleaner may fastidiously agree on the clean-up methodology to stop the code from making an attempt to entry a non-existing file or listing. Unfortunately, that strategy is sophisticated and error-prone. However, most of those issues are transitory, because the cleaner will finally create the right buildings.
Even extra often, the unsure nature of community programming signifies that some capabilities that summary a community name fail as a result of packets had been misplaced or corrupted.
A standard answer is to retry the failing code. This follow permits skipping previous transitional issues whereas nonetheless (finally) failing if the problem persists. Python has a number of libraries to make retrying simpler. This is a standard “finger exercise.”
Tenacity
One library that goes past a finger train and into helpful abstraction is tenacity. Install it with pip set up tenacity or depend upon it utilizing a dependencies = tenacity line in your pyproject.toml file.
Set up logging
A helpful built-in function of tenacity is help for logging. With error dealing with, seeing log particulars about retry makes an attempt is invaluable.
To permit the remaining examples show log messages, set up the logging library. In an actual program, the central entry level or a logging configuration plugin does this. Here’s a pattern:
import logging
logging.basicConfig(
stage=logging.INFO,
format="%(asctime)s:%(name)s:%(levelname)s:%(message)s",
)
TENACITY_LOGGER = logging.getLogger("Retrying")
Selective failure
To show the options of tenacity, it is useful to have a option to fail just a few instances earlier than lastly succeeding. Using unittest.mock is beneficial for this situation.
from unittest import mock
factor = mock.MagicMock(side_effect=[ValueError(), ValueError(), 3])
If you are new to unit testing, learn my article on mock.
Before displaying the ability of tenacity, have a look at what occurs whenever you implement retrying immediately inside a operate. Demonstrating this makes it simple to see the guide effort utilizing tenacity saves.
def useit(a_thing):
for i in vary(3):
attempt:
worth = a_thing()
besides ValueError:
TENACITY_LOGGER.information("Recovering")
proceed
else:
break
else:
elevate ValueError()
print("the value is", worth)
The operate may be known as with one thing that by no means fails:
>>> useit(lambda: 5)
the worth is 5
With the eventually-successful factor:
>>> useit(factor)
2023-03-29 17:00:42,774:Retrying:INFO:Recovering
2023-03-29 17:00:42,779:Retrying:INFO:Recovering
the worth is 3
Calling the operate with one thing that fails too many instances ends poorly:
attempt:
useit(mock.MagicMock(side_effect=[ValueError()] * 5 + [4]))
besides Exception as exc:
print("could not use it", repr(exc))
The outcome:
2023-03-29 17:00:46,763:Retrying:INFO:Recovering
2023-03-29 17:00:46,767:Retrying:INFO:Recovering
2023-03-29 17:00:46,770:Retrying:INFO:Recovering
couldn't use it ValueError()
Simple tenacity utilization
For probably the most half, the operate above was retrying code. The subsequent step is to have a decorator deal with the retrying logic:
import tenacity
my_retry=tenacity.retry(
cease=tenacity.stop_after_attempt(3),
after=tenacity.after_log(TENACITY_LOGGER, logging.WARNING),
)
Tenacity helps a specified variety of makes an attempt and logging after getting an exception.
The useit operate not has to care about retrying. Sometimes it is sensible for the operate to nonetheless think about retryability. Tenacity permits code to find out retryability by itself by elevating the particular exception TryAgain:
@my_retry
def useit(a_thing):
attempt:
worth = a_thing()
besides ValueError:
elevate tenacity.TryAgain()
print("the value is", worth)
Now when calling useit, it retries ValueError while not having customized retrying code:
useit(mock.MagicMock(side_effect=[ValueError(), ValueError(), 2]))
The output:
2023-03-29 17:12:19,074:Retrying:WARNING:Finished name to '__main__.useit' after 0.000(s), this was the first time calling it.
2023-03-29 17:12:19,080:Retrying:WARNING:Finished name to '__main__.useit' after 0.006(s), this was the 2nd time calling it.
the worth is 2
Configure the decorator
The decorator above is only a small pattern of what tenacity helps. Here’s a extra sophisticated decorator:
my_retry = tenacity.retry(
cease=tenacity.stop_after_attempt(3),
after=tenacity.after_log(TENACITY_LOGGER, logging.WARNING),
earlier than=tenacity.before_log(TENACITY_LOGGER, logging.WARNING),
retry=tenacity.retry_if_exception_type(ValueError),
wait=tenacity.wait_incrementing(1, 10, 2),
reraise=True
)
This is a extra sensible decorator instance with further parameters:
earlier than: Log earlier than calling the operateretry: Instead of solely retryingTryAgain, retry exceptions with the given standardswait: Wait between calls (that is particularly necessary if calling out to a service)reraise: If retrying failed, reraise the final try’s exception
Now that the decorator additionally specifies retryability, take away the code from useit:
@my_retry
def useit(a_thing):
worth = a_thing()
print("the value is", worth)
Here’s the way it works:
useit(mock.MagicMock(side_effect=[ValueError(), 5]))
The output:
2023-03-29 17:19:39,820:Retrying:WARNING:Starting name to '__main__.useit', that is the first time calling it.
2023-03-29 17:19:39,823:Retrying:WARNING:Finished name to '__main__.useit' after 0.003(s), this was the first time calling it.
2023-03-29 17:19:40,829:Retrying:WARNING:Starting name to '__main__.useit', that is the 2nd time calling it.
the worth is 5
Notice the time delay between the second and third log strains. It’s virtually precisely one second:
>>> useit(mock.MagicMock(side_effect=[5]))
2023-03-29 17:20:25,172:Retrying:WARNING:Starting name to '__main__.useit', that is the first time calling it.
the worth is 5
With extra element:
attempt:
useit(mock.MagicMock(side_effect=[ValueError("detailed reason")]*3))
besides Exception as exc:
print("retrying failed", repr(exc))
The output:
2023-03-29 17:21:22,884:Retrying:WARNING:Starting name to '__main__.useit', that is the first time calling it.
2023-03-29 17:21:22,888:Retrying:WARNING:Finished name to '__main__.useit' after 0.004(s), this was the first time calling it.
2023-03-29 17:21:23,892:Retrying:WARNING:Starting name to '__main__.useit', that is the 2nd time calling it.
2023-03-29 17:21:23,894:Retrying:WARNING:Finished name to '__main__.useit' after 1.010(s), this was the 2nd time calling it.
2023-03-29 17:21:25,896:Retrying:WARNING:Starting name to '__main__.useit', that is the third time calling it.
2023-03-29 17:21:25,899:Retrying:WARNING:Finished name to '__main__.useit' after 3.015(s), this was the third time calling it.
retrying failed ValueError('detailed cause')
Again, with KeyError as a substitute of ValueError:
attempt:
useit(mock.MagicMock(side_effect=[KeyError("detailed reason")]*3))
besides Exception as exc:
print("retrying failed", repr(exc))
The output:
2023-03-29 17:21:37,345:Retrying:WARNING:Starting name to '__main__.useit', that is the first time calling it.
retrying failed KeyError('detailed cause')
Separate the decorator from the controller
Often, comparable retrying parameters are wanted repeatedly. In these circumstances, it is best to create a retrying controller with the parameters:
my_retryer = tenacity.Retrying(
cease=tenacity.stop_after_attempt(3),
after=tenacity.after_log(TENACITY_LOGGER, logging.WARNING),
earlier than=tenacity.before_log(TENACITY_LOGGER, logging.WARNING),
retry=tenacity.retry_if_exception_type(ValueError),
wait=tenacity.wait_incrementing(1, 10, 2),
reraise=True
)
Decorate the operate with the retrying controller:
@my_retryer.wraps
def useit(a_thing):
worth = a_thing()
print("the value is", worth)
Run it:
>>> useit(mock.MagicMock(side_effect=[ValueError(), 5]))
2023-03-29 17:29:25,656:Retrying:WARNING:Starting name to '__main__.useit', that is the first time calling it.
2023-03-29 17:29:25,663:Retrying:WARNING:Finished name to '__main__.useit' after 0.008(s), this was the first time calling it.
2023-03-29 17:29:26,667:Retrying:WARNING:Starting name to '__main__.useit', that is the 2nd time calling it.
the worth is 5
This permits you to collect the statistics of the final name:
>>> my_retryer.statistics
{'start_time': 26782.847558759,
'attempt_number': 2,
'idle_for': 1.0,
'delay_since_first_attempt': 0.0075125470029888675}
Use these statistics to replace an inner statistics registry and combine together with your monitoring framework.
Extend tenacity
Many of the arguments to the decorator are objects. These objects may be objects of subclasses, permitting deep extensionability.
For instance, suppose the Fibonacci sequence ought to decide the wait instances. The twist is that the API for asking for wait time solely provides the try quantity, so the same old iterative approach of calculating Fibonacci is just not helpful.
One option to accomplish the aim is to make use of the closed formula:
Somewhat-known trick is skipping the subtraction in favor of rounding to the closest integer:
Which interprets to Python as:
int(((1 + sqrt(5))/2)**n / sqrt(5) + 0.5)
This can be utilized immediately in a Python operate:
from math import sqrt
def fib(n):
return int(((1 + sqrt(5))/2)**n / sqrt(5) + 0.5)
The Fibonacci sequence counts from 0 whereas the try numbers begin at 1, so a wait operate must compensate for that:
def wait_fib(rcs):
return fib(rcs.attempt_number - 1)
The operate may be handed immediately because the wait parameter:
@tenacity.retry(
cease=tenacity.stop_after_attempt(7),
after=tenacity.after_log(TENACITY_LOGGER, logging.WARNING),
wait=wait_fib,
)
def useit(factor):
print("value is", factor())
attempt:
useit(mock.MagicMock(side_effect=[tenacity.TryAgain()] * 7))
besides Exception as exc:
move
Try it out:
2023-03-29 18:03:52,783:Retrying:WARNING:Finished name to '__main__.useit' after 0.000(s), this was the first time calling it.
2023-03-29 18:03:52,787:Retrying:WARNING:Finished name to '__main__.useit' after 0.004(s), this was the 2nd time calling it.
2023-03-29 18:03:53,789:Retrying:WARNING:Finished name to '__main__.useit' after 1.006(s), this was the third time calling it.
2023-03-29 18:03:54,793:Retrying:WARNING:Finished name to '__main__.useit' after 2.009(s), this was the 4th time calling it.
2023-03-29 18:03:56,797:Retrying:WARNING:Finished name to '__main__.useit' after 4.014(s), this was the fifth time calling it.
2023-03-29 18:03:59,800:Retrying:WARNING:Finished name to '__main__.useit' after 7.017(s), this was the sixth time calling it.
2023-03-29 18:04:04,806:Retrying:WARNING:Finished name to '__main__.useit' after 12.023(s), this was the seventh time calling it.
Subtract subsequent numbers from the “after” time and spherical to see the Fibonacci sequence:
intervals = [
0.000,
0.004,
1.006,
2.009,
4.014,
7.017,
12.023,
]
for x, y in zip(intervals[:-1], intervals[1:]):
print(int(y-x), finish=" ")
Does it work? Yes, precisely as anticipated:
0 1 1 2 3 5
Wrap up
Writing ad-hoc retry code is usually a enjoyable distraction. For production-grade code, a better option is a confirmed library like tenacity. The tenacity library is configurable and extendable, and it’ll seemingly meet your wants.
