Tubthumper: Helping you get up ... again!

What's in a name?

Tubthumper is a Python package of retry utilities named after the English anarcho-communist rock band Chumbawamba's 1997 hit Tubthumping. Yes, really.

I get knocked down, but I get up again. 🎶
You're never gonna keep me down. 🎶
I get knocked down, but I get up again. 🎶
You're never gonna keep me down... 🎶

Getting Started

Installation

tubthumper is a pip-installable package hosted on PyPI. Getting started is as easy as:

$ pip install tubthumper

tubthumper requires Python 3.9 or greater. For Python 3.10 or greater, it has no external dependencies, i.e. standard library only, but earlier versions require typing-extensions.

Usage

Import tubthumper's useful bits:

>>> from tubthumper import retry, retry_decorator, retry_factory

Call a function with retry and jittered exponential backoff:

>>> retry(get_ip, exceptions=ConnectionError)
WARNING: Function threw exception below on try 1, retrying in 0.844422 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
{'ip': '8.8.8.8'}

Call that same function with positional and keyword arguments, e.g. retry get_ip(42, "test", dev=True):

>>> retry(get_ip,
...     args=(42, "test"), kwargs={"dev": True},
...     exceptions=ConnectionError)
WARNING: Function threw exception below on try 1, retrying in 0.420572 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
{'ip': '8.8.8.8'}

Bake retry behavior into your function with a decorator:

>>> @retry_decorator(exceptions=ConnectionError)
... def get_ip_retry():
...     return requests.get("http://ip.jsontest.com").json()
>>> get_ip_retry()
WARNING: Function threw exception below on try 1, retrying in 0.511275 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
{'ip': '8.8.8.8'}

Create a new function with retry behavior from an existing one:

>>> get_ip_retry = retry_factory(get_ip, exceptions=ConnectionError)
>>> get_ip_retry()
WARNING: Function threw exception below on try 1, retrying in 0.783799 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
{'ip': '8.8.8.8'}

Customization

While tubthumper ships with a set of sensible defaults, its retry behavior is fully customizable.

Exceptions

Because overbroad except clauses are the most diabolical Python antipattern, there is no sensible default for what exception or exceptions to catch and retry. Thus, every tubthumper interface has a required exceptions keyword-only argument, which takes an exception or tuple of exceptions to catch and retry on, i.e. a sensible lack of a default.

>>> retry(get_ip, exceptions=ConnectionError)
WARNING: Function threw exception below on try 1, retrying in 0.476597 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
{'ip': '8.8.8.8'}
>>> retry(get_ip, exceptions=(KeyError, ConnectionError))
WARNING: Function threw exception below on try 1, retrying in 0.908113 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
{'ip': '8.8.8.8'}

By default, tubthumper raises a tubthumper.RetryError exception when all retries have been exhausted:

>>> retry(lambda: 1/0, retry_limit=0, exceptions=ZeroDivisionError)
Traceback (most recent call last):
  ...
tubthumper._retry_factory.RetryError: Retry limit 0 reached

You can override this behavior using the reraise flag to reraise the original exception in place of RetryError:

>>> retry(lambda: 1/0, retry_limit=0, reraise=True, exceptions=ZeroDivisionError)
Traceback (most recent call last):
  ...
ZeroDivisionError: division by zero

Retry Limits

By default, tubthumper will retry endlessly, but you have two means of limiting retry behavior. As shown previously, to limit the number of retries attempted, use the retry_limit keyword-only argument:

>>> retry(lambda: 1/0, retry_limit=10, exceptions=ZeroDivisionError)
...  # Warning logs for each failed call
Traceback (most recent call last):
  ...
tubthumper._retry_factory.RetryError: Retry limit 10 reached

Alternatively, you can use the time_limit keyword-only argument to prevent retry attempts after a certain duration:

>>> retry(lambda: 1/0, time_limit=60, exceptions=ZeroDivisionError)
...  # Warning logs for each failed call
Traceback (most recent call last):
  ...
tubthumper._retry_factory.RetryError: Time limit 60 exceeded

Backoff timing

By default, the backoff duration doubles with each retry, starting off at one second. As well, each backoff period is jittered, i.e. scaled by a uniformly distributed random number on the [0.0, 1.0) interval. You can disable jittering using the jitter keyword-only argument:

>>> retry(get_ip, jitter=False, exceptions=ConnectionError)
WARNING: Function threw exception below on try 1, retrying in 1 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
{'ip': '8.8.8.8'}

You can set the initial backoff duration using the init_backoff keyword-only argument:

>>> retry(get_ip, jitter=False, init_backoff=10, exceptions=ConnectionError)
WARNING: Function threw exception below on try 1, retrying in 10 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
{'ip': '8.8.8.8'}

Finally, you can set the factor by which each successive backoff duration is scaled using the exponential keyword-only argument:

>>> retry(get_ip, jitter=False, exponential=3, exceptions=ConnectionError)
WARNING: Function threw exception below on try 1, retrying in 1 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
WARNING: Function threw exception below on try 2, retrying in 3 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
{'ip': '8.8.8.8'}

Logging

By default, tubthumper logs each caught exception at the logging.WARNING level using a logger named tubthumper, i.e. logging.getLogger("tubthumper"). As described in the Python logging tutorial, for this default logger, "events of severity WARNING and greater will be printed to sys.stderr" if no further logging is configured.

You can set the logging level using the log_level keyword-only argument:

>>> retry(get_ip, log_level=logging.DEBUG, exceptions=ConnectionError) # No warnings
{'ip': '8.8.8.8'}

You can provide your own logger using the logger keyword-only argument. This logger's log method will be called like so:

logger.log(log_level, "Function threw...", exc_info=True)

Features

Compatible with methods

tubthumper's various interfaces are compatible with methods, including classmethods and staticmethods:

>>> class Class:
...     @retry_decorator(exceptions=ConnectionError)
...     def get_ip(self):
...         return requests.get("http://ip.jsontest.com").json()
...
>>> Class().get_ip()
WARNING: Function threw exception below on try 1, retrying in 0.719705 seconds
Traceback (most recent call last):
  ...
requests.exceptions.ConnectionError: http://ip.jsontest.com
{'ip': '8.8.8.8'}

Signature preserving

tubthumper's various interfaces preserve the relevant dunder attributes of your function:

>>> @retry_decorator(exceptions=ConnectionError)
... def func(one: bool, two: float = 3.0) -> complex:
...     """This is a docstring"""
...
>>> func.__name__
'func'
>>> func.__qualname__
'func'
>>> func.__module__
'__main__'
>>> func.__doc__
'This is a docstring'
>>> func.__annotations__
{'one': <class 'bool'>, 'two': <class 'float'>, 'return': <class 'complex'>}

tubthumper also preserves the inspect module's function signature, and is* functions:

>>> import inspect
>>> inspect.signature(func)
<Signature (one: bool, two: float = 3.0) -> complex>
>>> inspect.isfunction(func)
True
>>> inspect.isroutine(func)
True
>>> inspect.ismethod(Class().get_ip)
True

Async support

tubthumper's various interfaces support coroutine functions, including generator-based coroutines, awaiting them while using async.sleep between awaits:

>>> @retry_decorator(exceptions=ConnectionError)
... async def get_ip():
...     return requests.get("http://ip.jsontest.com").json()
...
>>> inspect.iscoroutinefunction(get_ip)
True

Fully type annotated

tubthumper's various interfaces are fully type annotated, passing pyright.

100% Test Coverage

tubthumper achieves 100% test coverage across three supported operating systems (Windows, MacOS, & Linux). You can find the coverage report on Codecov.