Last active
April 5, 2020 09:33
-
-
Save rednafi/212029ac79067650796766eb8167dcca to your computer and use it in GitHub Desktop.
Retry decorator with exponential backoff
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
import time | |
from functools import partial, wraps | |
def retry(func=None, exception=Exception, n_tries=5, delay=5, backoff=1, logger=False): | |
"""Retry decorator with exponential backoff. | |
Parameters | |
---------- | |
func : typing.Callable, optional | |
Callable on which the decorator is applied, by default None | |
exception : Exception or tuple of Exceptions, optional | |
Exception(s) that invoke retry, by default Exception | |
n_tries : int, optional | |
Number of tries before giving up, by default 5 | |
delay : int, optional | |
Initial delay between retries in seconds, by default 5 | |
backoff : int, optional | |
Backoff multiplier e.g. value of 2 will double the delay, by default 1 | |
logger : bool, optional | |
Option to log or print, by default False | |
Returns | |
------- | |
typing.Callable | |
Decorated callable that calls itself when exception(s) occur. | |
Examples | |
-------- | |
>>> import random | |
>>> @retry(exception=Exception, n_tries=4) | |
... def test_random(text): | |
... x = random.random() | |
... if x < 0.5: | |
... raise Exception("Fail") | |
... else: | |
... print("Success: ", text) | |
>>> test_random("It works!") | |
""" | |
if func is None: | |
return partial( | |
retry, | |
exception=exception, | |
n_tries=n_tries, | |
delay=delay, | |
backoff=backoff, | |
logger=logger, | |
) | |
@wraps(func) | |
def wrapper(*args, **kwargs): | |
ntries, ndelay = n_tries, delay | |
while ntries > 1: | |
try: | |
return func(*args, **kwargs) | |
except exception as e: | |
msg = f"{str(e)}, Retrying in {ndelay} seconds..." | |
if logger: | |
logging.warning(msg) | |
else: | |
print(msg) | |
time.sleep(ndelay) | |
ntries -= 1 | |
ndelay *= backoff | |
return func(*args, **kwargs) | |
return wrapper |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment