Last active
May 10, 2019 08:15
-
-
Save ceoro9/240ba1f791fe00b2e89bd884f282e1d0 to your computer and use it in GitHub Desktop.
Call method with retries
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
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