Retry decorator in Python (try again on failure)

Some operations sometimes fail for reasons outside your program, such as a slow server. A retry decorator calls the function again a few times before giving up, so you do not have to write that loop around every call.

The program

python

import functools
import time


def retry(times=3, delay=1.0, exceptions=(Exception,)):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as error:
                    print(f"Attempt {attempt} failed: {error}")
                    if attempt == times:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator


calls = 0


# this function fails twice and then works, so the output is the same every run
@retry(times=3, delay=0.05)
def fetch_data():
    global calls
    calls += 1
    if calls < 3:
        raise ConnectionError("server did not answer")
    return "data received"


print(fetch_data())

+ setup added so this can run · defines args, kwargs

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

args = _AutoMock('args')
kwargs = _AutoMock('kwargs')

output

Attempt 1 failed: server did not answer
Attempt 2 failed: server did not answer
data received

How it works

retry(times=3, delay=0.05) is called first and returns the real decorator. That extra layer is what lets a decorator take arguments.
wrapper tries the function up to times times. If the call works, return ends the loop at once.
except exceptions catches only the error types you list. The default, (Exception,), catches almost everything. In real code, list the errors that may not happen again on the next attempt, such as (ConnectionError, TimeoutError), so that a bug like a TypeError is not retried.
On the last attempt, a bare raise raises the same error again, so the caller still sees what went wrong. The program also does not wait after the last attempt, because there is nothing left to wait for.
functools.wraps(func) copies the name and docstring of the original function onto wrapper.

When every attempt fails

python

import functools
import time


def retry(times=3, delay=1.0, exceptions=(Exception,)):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as error:
                    print(f"Attempt {attempt} failed: {error}")
                    if attempt == times:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator


@retry(times=3, delay=0.05, exceptions=(ConnectionError,))
def always_down():
    raise ConnectionError("server did not answer")


try:
    always_down()
except ConnectionError as error:
    print("Gave up:", error)

+ setup added so this can run · defines args, kwargs

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

args = _AutoMock('args')
kwargs = _AutoMock('kwargs')

output

Attempt 1 failed: server did not answer
Attempt 2 failed: server did not answer
Attempt 3 failed: server did not answer
Gave up: server did not answer

Many real systems double the delay after each failure (1 second, then 2, then 4). This is called exponential backoff, and it gives a busy server time to recover. To add it here, start wrapper with wait = delay, call time.sleep(wait) instead of time.sleep(delay), and put wait *= 2 after it.