in src/helper.py [0:0]
def retry(max_attempts: int = 5, delay: int = 3, error_code=None, error_message=None):
"""Used as a decorator to retry any definition based on error code or error message, typically used due
to lower then normal API limits.
Args:
max_attempts (int): Max number of retries
delay (int): Duration to wait before another retry
error_code (list) [optional]: Error code to search for in error
error_message (str) [optional]: Error message to search for to retry
Returns:
:obj:`json`: Returns json object (dict) of the user parameters
"""
if not error_code:
error_code = ['TooManyRequestsException']
def retry_decorator(function):
@wraps(function)
def wrapper(*args, **kwargs):
last_exception = "default exception"
m_attempts = max_attempts # create new memory allocation
while m_attempts > 1:
try:
logger.debug("**** [retry] args:%s", args)
logger.debug("**** [retry] kwargs:%s", kwargs)
return function(*args, **kwargs)
except Exception as err:
if hasattr(err, 'response'):
logger.debug("**** [retry] ErrorCode:%s", err.response['Error']['Code'])
logger.warning(err)
if error_message and re.search(error_message, str(err)):
logger.warning(
"Definition failed:%s with (%s) message, trying again "
"in %s seconds...", function.__name__, error_message, delay
)
time.sleep(delay)
m_attempts -= 1
last_exception = err
elif hasattr(err, 'response'):
if err.response['Error']['Code'] in error_code:
logger.warning(
"Definition failed:%s with (%s) "
"error code, trying again in %s seconds...",
function.__name__, err.response['Error']['Code'], delay
)
time.sleep(delay)
m_attempts -= 1
last_exception = err
else:
logger.warning("Error wasn't found in retry raising error:%s", err)
raise err
logger.error("Was not successfully able to complete the request after %s attempts", max_attempts)
raise last_exception
return wrapper
return retry_decorator