Python Retry Decorator

retry.py

# Copyright 2021 Fabian Bosler
# from: https://towardsdatascience.com/are-you-using-python-with-apis-learn-how-to-use-a-retry-decorator-27b6734c3e6
# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy,
# modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom
# the Software is furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
# Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
# THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS
# OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

from functools import wraps
import time
import random
import sys
from typing import Any, Callable
from shared.app_settings import AppSettings
from shared.types import LoggingProperties
from shared.logging_helper import get_logger, LoggingHelper

app_settings = AppSettings()
logger = get_logger(
    "[breathe.fn_workflow_interface]",
    log_properties=LoggingProperties(
        site_name=app_settings.SITE_NAME,
        environment=app_settings.ENVIRONMENT,
        stamp=app_settings.STAMP,
    ),
)
logger.info(
    f"functions running under Python {sys.version}",
)


def retry(exceptions: Exception, total_tries: int = 4, initial_wait: float = 0.5, backoff_factor: int = 2, logger: LoggingHelper = logger):
    """
    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: Callable[..., Any]):
        @wraps(f)
        def func_with_retries(*args: str, **kwargs: str):
            _tries, _delay = total_tries + 1, initial_wait
            while _tries > 1:
                try:
                    logger.info(f"{total_tries + 2 - _tries}. try:")
                    return f(*args, **kwargs)
                except Exception 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}"
                        )
                        logger.exception(msg)
                        raise e
                    msg = str(
                        f"Function: {f.__name__}\n"
                        f"Exception: {e}\n"
                        f"Retrying in {_delay} seconds!, args: {print_args}, kwargs: {kwargs}\n"
                    )
                    logger.warning(msg)
                    time.sleep(_delay)
                    _delay *= backoff_factor

        return func_with_retries

    return retry_decorator


def test_func(*args: str, **kwargs: str):
    rnd = random.random()
    if rnd < 0.2:
        raise ConnectionAbortedError("Connection was aborted :(")
    elif rnd < 0.4:
        raise ConnectionRefusedError("Connection was refused :/")
    elif rnd < 0.8:
        raise ConnectionResetError("Guess the connection was reset")
    else:
        return "Yay!!"


if __name__ == "__main__":
    # wrapper = retry((ConnectionAbortedError), tries=3, delay=.2, backoff=1, logger=logger)
    # wrapped_test_func = wrapper(test_func)
    # print(wrapped_test_func('hi', 'bye', hi='ciao'))

    wrapper_all_exceptions = retry(Exception, total_tries=2)
    wrapped_test_func = wrapper_all_exceptions(test_func)
    print(wrapped_test_func("hi", "bye", hi="ciao"))