Python (Auto) Retry - Automatically retry your calls using optional delay - asyncio compatible

Retry.py

import asyncio
import functools
import time


__all__ = ['Retry']


class Retry(object):
    """

    Python Retry - Automatically retry your calls using optional delay [detects blocking or non-blocking]

    This wrapper/decorator will retry the wrapped function if:
        - there is no return value
        - an exception occured

    """

    def __init__(self, retries=5, delay=2, debug=True, ignore_exc_types=()):
        """
        Will retry the wrapped function if there is no return value or an exception is raised

        If the decorated function is running inside an asyncio event loop, the delay is non-blocking

        :param retries (int) : number of retries before giving up
        :param delay (float) : delay in seconds between each retry attempt
        :param debug (bool)  : print retry attempts to console
        :param ignore_exc_types (iterable) : iterable of exception classes that not trigger retrying

        """
        self.retries = retries
        self.delay = delay
        self._debug = debug
        self.ignore_exc_types = ignore_exc_types
        try:
            self._async = asyncio.get_running_loop()
        except:
            self._async = False

    def __call__(self, fn):

        if not self._async:

            @functools.wraps(fn)
            def decorated(*a, **k):
                result = None
                last_exception = None
                for attempt in range(self.retries):
                    try:
                        result = fn(*a, **k)
                        if result:
                            return result
                    except self.ignore_exc_types as e:
                        last_exception = e
                    if self._debug:
                        print('{} retrying {} - in {} seconds - attempt {}/{}'.format(
                            self.__class__.__name__, fn.__name__, self.delay, attempt + 1, self.retries))
                    time.sleep(self.delay)
                if last_exception is not None:
                    raise type(last_exception) from last_exception
                return result

            return decorated

        elif self._async:

            @functools.wraps(fn)
            async def decorated(*a, **k):
                result = None
                last_exception = None
                for attempt in range(self.retries):
                    try:
                        result = fn(*a, **k)
                        if result:
                            return result
                    except self.ignore_exc_types as e:
                        last_exception = e
                    if self._debug:
                        print('{} retrying {} - in {} seconds - attempt {}/{}'.format(
                            self.__class__.__name__, fn.__name__, self.delay, attempt + 1, self.retries))
                    await asyncio.sleep(self.delay)
                if last_exception is not None:
                    raise type(last_exception) from last_exception
                return result

            return decorated