Last active
February 22, 2022 04:45
-
-
Save favtuts/b6341a301e112d4f782440ee81660c2d to your computer and use it in GitHub Desktop.
Python retry decorator implements simple retry pattern with exponential backoff algorithm. More details on blog post: https://www.favtuts.com/retry-decorator-as-python-utility-to-protect-unexpected-one-off-exceptions-while-calling-apis/
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
from asyncio.log import logger | |
from functools import wraps | |
import time | |
import logging | |
import random | |
logger = logging.getLogger(__name__) | |
def log(msg, logger = None): | |
if logger: | |
logger.warning(msg) | |
else: | |
print(msg) | |
def retry(exceptions, total_tries=4, initial_wait=0.5, backoff_factor=2, logger=None): | |
""" | |
calling the decorated function applying an exponential backoff. | |
Args: | |
exceptions: Exception(s) that trigger a retry, can be a tuple | |
total_tries: Total tries | |
initial_wait: Time to first retry | |
backoff_factor: Backoff multiplier (e.g. value of 2 will double the delay each retry). | |
logger: logger to be used, if none specified print | |
""" | |
def retry_decorator(f): | |
@wraps(f) | |
def func_with_retries(*args, **kwargs): | |
_tries, _delay = total_tries + 1, initial_wait | |
while _tries > 1: | |
try: | |
log(f'{total_tries + 2 - _tries}. try:', logger) | |
return f(*args, **kwargs) | |
except exceptions as e: | |
_tries -= 1 | |
print_args = args if args else 'no args' | |
if _tries == 1: | |
msg = str(f'Function: {f.__name__}\n' | |
f'Failed despite best efforts after {total_tries} tries.\n' | |
f'args: {print_args}, kwargs: {kwargs}') | |
log(msg, logger) | |
raise | |
msg = str(f'Function: {f.__name__}\n' | |
f'Exception: {e}\n' | |
f'Retrying in {_delay} seconds!, args: {print_args}, kwargs: {kwargs}\n') | |
log(msg, logger) | |
time.sleep(_delay) | |
_delay *= backoff_factor | |
return func_with_retries | |
return retry_decorator | |
@retry(Exception, total_tries=4, logger=logger) | |
def test_func(*args, **kwargs): | |
rnd = random.random() | |
if rnd < .2: | |
raise ConnectionAbortedError('Connection was aborted :(') | |
elif rnd < .4: | |
raise ConnectionRefusedError('Connection was refused :/') | |
elif rnd < .8: | |
raise ConnectionResetError('Guess the connection was reset') | |
else: | |
return 'Yay!!' | |
@retry(Exception, total_tries=10, logger=logger) | |
def test_func_with_decor(*args, **kwargs): | |
rnd = random.random() | |
if rnd < .2: | |
raise ConnectionAbortedError('Connection was aborted :(') | |
elif rnd < .4: | |
raise ConnectionRefusedError('Connection was refused :/') | |
elif rnd < .6: | |
raise ConnectionResetError('Guess the connection was reset') | |
elif rnd < .8: | |
raise TimeoutError('This took too long') | |
else: | |
return 'Yay!!' | |
if __name__ == '__main__': | |
# TEST wrapper for specific Exception | |
# wrapper = retry((ConnectionAbortedError), total_tries=3, logger=logger) | |
# wrapped_test_func = wrapper(test_func) | |
# print(wrapped_test_func('hi', 'bye', hi='ciao')) | |
# TEST wrapper for all Exceptions | |
# wrapper_all_exceptions = retry(Exception, total_tries=10, logger=logger) | |
# wrapped_test_func = wrapper_all_exceptions(test_func) | |
# print(wrapped_test_func('hi', 'bye', hi='ciao')) | |
# TEST decorator for all Exceptions | |
print(test_func_with_decor('hi', 'bye', hi='ciao')) | |
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
def first_func(x): | |
return x**2 | |
def second_func(x): | |
return x - 2 | |
# print(first_func('2')) # output: unsupported operand type(s) for ** or pow(): 'str' and 'int' | |
# print(second_func('2')) # output: unsupported operand type(s) for ** or pow(): 'str' and 'int' | |
def convert_to_numeric(func): | |
# define a function within the outer function | |
def new_func(x): | |
return func(float(x)) | |
# return the newly defined function | |
return new_func | |
#new_first_func = convert_to_numeric(first_func) | |
#print(new_first_func('2')) # output: 4.0 | |
#print(convert_to_numeric(second_func)('2')) # output: 0.0 | |
# third function with decorator | |
@convert_to_numeric | |
def third_func(x): | |
return x**2 + x - 2 | |
print(third_func('2')) # output : 2^2 + 2 - 2 = 4.0 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment