Last active
April 28, 2020 11:23
-
-
Save dtheodor/88b6e070ddfc8ddc77074dcec16b882d to your computer and use it in GitHub Desktop.
Python retry decorator
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
from decorator import decorator | |
def retry(retries, delay, backoff, should_retry_fn, jitter=0, max_delay=None, | |
on_retry_fn=None, logger=_logger, label=None): | |
@decorator | |
def _retry(f, *args, **kwargs): | |
retries_left, current_delay = retries, delay | |
while retries_left: | |
try: | |
return f(*args, **kwargs) | |
except Exception as ex: | |
if not should_retry_fn(ex, *args, **kwargs): | |
raise | |
attempt_num = retries - retries_left + 1 | |
retries_left -= 1 | |
attempt_error_msg = \ | |
u'{label} Attempt {num} of {total}: {exc_type} {exc}'.format( | |
label=u'{}: '.format(label) if label else u'', | |
num=attempt_num, | |
total=retries, | |
exc_type=type(ex).__name__, | |
exc=ex) | |
if not retries_left: | |
logger.info(attempt_error_msg) | |
raise | |
if on_retry_fn: | |
on_retry_fn(*args, **kwargs) | |
logger.debug( | |
u'%s, retrying in %s seconds...', | |
attempt_error_msg, current_delay) | |
logger.debug('Retryable error traceback:', exc_info=True) | |
time.sleep(current_delay) | |
current_delay *= backoff | |
if isinstance(jitter, tuple): | |
current_delay += random.uniform(*jitter) | |
else: | |
current_delay += jitter | |
if max_delay is not None: | |
current_delay = min(current_delay, max_delay) | |
return _retry | |
@retry(should_retry_fn=lambda e: isinstance(e, RetryableError)) | |
def do(): | |
# do work that may raise a RetryableError |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment