Skip to content

Instantly share code, notes, and snippets.

@dtheodor
Last active April 28, 2020 11:23
Show Gist options
  • Save dtheodor/88b6e070ddfc8ddc77074dcec16b882d to your computer and use it in GitHub Desktop.
Save dtheodor/88b6e070ddfc8ddc77074dcec16b882d to your computer and use it in GitHub Desktop.
Python retry decorator
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