Skip to content

Instantly share code, notes, and snippets.

@mxsrc
Created June 5, 2025 09:49
Show Gist options
  • Save mxsrc/3c0a930806b7de049505c5e515344535 to your computer and use it in GitHub Desktop.
Save mxsrc/3c0a930806b7de049505c5e515344535 to your computer and use it in GitHub Desktop.
Retry helper for Python
import logging
import random
import time
from functools import wraps
logging.basicConfig(level=logging.DEBUG, format='%(message)s')
logger = logging.getLogger()
class RetriesExceededError(Exception):
pass
def retry(
func,
predicate=None,
retries=0, delay=0.,
exceptions=(), expected_exceptions=(),
linear_backoff=0., multiplicative_backoff=1., exponential_backoff=1.,
):
@wraps(func)
def wrapper(*args, **kwargs):
d = delay
for i in range(retries + 1):
try:
result = func(*args, **kwargs)
if expected_exceptions != ():
pass # We are waiting for a specific exception
elif predicate is not None and not predicate(result):
logger.debug(f'{func.__name__}() returned bad value')
else:
return result
except expected_exceptions:
return
except exceptions:
logger.debug(f'{func.__name__}() failed, retrying (attempt {i})...', exc_info=True)
logger.info(f'Retrying {func.__name__}() (attempt {i})...')
time.sleep(d)
d = linear_backoff + multiplicative_backoff * delay ** exponential_backoff
raise RetriesExceededError(f'Retrying {func.__name__}() failed ({retries + 1} attempts)')
return wrapper
def f(cutoff):
if random.random() < cutoff:
print('Failure...')
raise ValueError()
print('Success!')
if __name__ == '__main__':
retry(f)(.5)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment