Skip to content

Instantly share code, notes, and snippets.

@ceoro9
Last active May 10, 2019 08:15
Show Gist options
  • Save ceoro9/240ba1f791fe00b2e89bd884f282e1d0 to your computer and use it in GitHub Desktop.
Save ceoro9/240ba1f791fe00b2e89bd884f282e1d0 to your computer and use it in GitHub Desktop.
Call method with retries
import time
class CustomException(Exception):
pass
def retry(max_attempts=1,
should_retry=lambda _e: True,
retryable_exceptions=tuple(),
delay_time=1,
):
"""Retrier
Retry calling function
Keyword Arguments:
max_attempts {number} -- Maximum number of retries (default: {1})
should_retry {function} -- Function that accepts exception and determine should
retrier continue working (default: {lambda _e: True})
retryable_exceptions {[type]} -- Exceptions that should be retried (default: {tuple()})
delay_time {number} -- Time between attempts (default: {1 second})
"""
def wrap_retry(func):
def handle_func(*args, **kwargs):
attemtps = 0
while True:
try:
response = func(*args, **kwargs)
except retryable_exceptions as e:
attemtps += 1
if attemtps >= max_attempts or not should_retry(e):
raise e
time.sleep(delay_time)
continue
return response
return handle_func
return wrap_retry
@retry(max_attempts=3, retryable_exceptions=(CustomException,))
def process():
s = input("S = ")
if s == '1':
print('raising retryable exception.')
raise CustomException('1')
elif s == '2':
print('raising unretryable exception')
raise Exception('2')
else:
print('Success')
if __name__ == '__main__':
process()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment