Skip to content

Instantly share code, notes, and snippets.

@favtuts
Last active February 22, 2022 04:45
Show Gist options
  • Save favtuts/b6341a301e112d4f782440ee81660c2d to your computer and use it in GitHub Desktop.
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/
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'))
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