Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import functools
- import time
- import logging
- logging.basicConfig(level=logging.DEBUG, format='%(message)s')
- logger = logging.getLogger(__name__)
- def retry(predicate=None, max_tries=3, max_delay=60,
- delay_s=1, exponential=False):
- if predicate is None:
- predicate = bool
- def decorator(f):
- def it():
- def factor(p):
- if exponential:
- return pow(2, p)
- return 1
- for p in range(max_tries):
- yield min(factor(p) * delay_s, max_delay)
- @functools.wraps(f)
- def wrapper(*args, **kwargs):
- last_e = None
- for delay in it():
- try:
- result = f(*args, **kwargs)
- except BaseException as e:
- last_e = e
- logger.exception('error in retry decorator!')
- else:
- if predicate(result):
- return result
- logger.debug('retrying after {} seconds'.format(delay))
- time.sleep(delay)
- if last_e:
- raise last_e
- raise Exception('No more retries!')
- return wrapper
- return decorator
- if __name__ == '__main__':
- def flaky(n_errors=0, max_tries=5,
- predicate=lambda _: True, exponential=True):
- n = {'errors': n_errors}
- @retry(max_tries=max_tries,
- predicate=predicate, exponential=exponential)
- def f():
- if n['errors'] <= 0:
- return 'success!'
- n['errors'] -= 1
- raise Exception('boom!')
- return f
- # within retry limit
- f = flaky(n_errors=3, max_tries=5)
- print(f())
- # no more retries
- f = flaky(n_errors=3, max_tries=3)
- print(f())
Add Comment
Please, Sign In to add comment