Skip to main content

Small and versatile library to retry failed operations using different backoff strategies

Project description

riprova |Build Status| |PyPI| |Coverage Status| |Documentation Status| |Quality| |Versions|

``riprova`` (meaning ``retry`` in Italian) is a small, general-purpose and versatile `Python`_ library
that provides retry mechanisms with multiple backoff strategies for any sort of failed operations.

It's domain agnostic, highly customizable, extensible and provides a minimal API that's easy to instrument in any code base via decorators, context managers or raw API consumption.

For a brief introduction about backoff mechanisms for potential failed operations, `read this article`_.


- Retry decorator for simple and idiomatic consumption.
- Simple Pythonic programmatic interface.
- Maximum retry timeout support.
- Supports error `whitelisting`_ and `blacklisting`_.
- Supports custom `error evaluation`_ retry logic (useful to retry only in specific cases).
- Automatically retry operations on raised exceptions.
- Supports `asynchronous coroutines`_ with both ``async/await`` and ``yield from`` syntax.
- Configurable maximum number of retry attempts.
- Highly configurable supporting max retries, timeouts or retry notifier callback.
- Built-in backoff strategies: constant, `fibonacci`_ and `exponential`_ backoffs.
- Supports sync/async context managers.
- Pluggable custom backoff strategies.
- Lightweight library with almost zero embedding cost.
- Works with Python +2.6, 3.0+ and PyPy.

Backoff strategies

List of built-in backoff strategies.

- `Constant backoff`_
- `Fibonacci backoff`_
- `Exponential backoff`_

You can also implement your own one easily.
See `ConstantBackoff`_ for an implementation reference.


Using ``pip`` package manager (requires pip 1.9+. Upgrade it running: ``pip install -U pip``):

.. code-block:: bash

pip install -U riprova

Or install the latest sources from Github:

.. code-block:: bash

pip install -e git+git://


- riprova.retry_
- riprova.Retrier_
- riprova.AsyncRetrier_
- riprova.Backoff_
- riprova.ConstantBackoff_
- riprova.FibonacciBackoff_
- riprova.ExponentialBackoff_
- riprova.ErrorWhitelist_
- riprova.ErrorBlacklist_
- riprova.add_whitelist_error_
- riprova.RetryError_
- riprova.RetryTimeoutError_
- riprova.MaxRetriesExceeded_
- riprova.NotRetriableError_

.. _riprova.retry:
.. _riprova.Retrier:
.. _riprova.AsyncRetrier:
.. _riprova.Backoff:
.. _riprova.ConstantBackoff:
.. _riprova.FibonacciBackoff:
.. _riprova.ExponentialBackoff:
.. _riprova.ErrorWhitelist:
.. _riprova.ErrorBlacklist:
.. _riprova.add_whitelist_error:
.. _riprova.RetryError:
.. _riprova.RetryTimeoutError:
.. _riprova.MaxRetriesExceeded:
.. _riprova.NotRetriableError:


You can see more featured examples from the `documentation` site.

**Basic usage examples**:

.. code-block:: python

import riprova

def task():
"""Retry operation if it fails with constant backoff (default)"""

def task():
"""Retry operation if it fails with custom max number of retry attempts"""

def task():
"""Retry operation if it fails using exponential backoff"""

def task():
"""Raises a TimeoutError if the retry loop exceeds from 10 seconds"""

def on_retry(err, next_try):
print('Operation error: {}'.format(err))
print('Next try in: {}ms'.format(next_try))

def task():
"""Subscribe via function callback to every retry attempt"""

def evaluator(response):
# Force retry operation if not a valid response
if response.status >= 400:
raise RuntimeError('invalid response status') # or simple return True
# Otherwise return False, meaning no retry
return False

def task():
"""Use a custom evaluator function to determine if the operation failed or not"""

async def task():
"""Asynchronous coroutines are also supported :)"""

**Retry failed HTTP requests**:

.. code-block:: python

import pook
import requests
from riprova import retry

# Define HTTP mocks to simulate failed requests
pook.get('').times(1).reply(200).json({'hello': 'world'})

# Retry evaluator function used to determine if the operated failed or not
def evaluator(response):
if response != 200:
return Exception('failed request') # you can also simply return True
return False

# On retry even subscriptor
def on_retry(err, next_try):
print('Operation error {}'.format(err))
print('Next try in {}ms'.format(next_try))

# Register retriable operation
@retry(evaluator=evaluator, on_retry=on_retry)
def fetch(url):
return requests.get(url)

# Run task that might fail


MIT - Tomas Aparicio

.. _exponential:
.. _fibonacci:
.. _asyncio:
.. _Python:
.. _annotated API reference:
.. _async/await:
.. _yield from:
.. _documentation:
.. _read this article:
.. _Constant backoff:
.. _Fibonacci backoff:
.. _Exponential backoff:
.. _ConstantBackoff:
.. _whitelisting:
.. _blacklisting:
.. _error evaluation:
.. _asynchronous coroutines:

.. |Build Status| image::
.. |PyPI| image::
.. |Coverage Status| image::
.. |Documentation Status| image::
.. |Quality| image::
.. |Stability| image::
.. |Versions| image::


v0.2.5 / 2018-03-21

* Merge pull request #15 from jstasiak/allow-newer-six
* Allow newer six
* feat(History): update changes

v0.2.5 / 2018-03-21

* Merge pull request #15 from jstasiak/allow-newer-six
* Allow newer six
* feat(History): update changes

v0.2.4 / 2018-03-20

* merge(#14): Allow subsecond maxtimes for ExponentialBackoff

v0.2.3 / 2017-01-13

* refactor(retry): remove unnecessary partial function
* fix(retry): rename keyword param for partial application
* feat(docs): improve description
* refactor(Makefile): update publish task

v0.2.2 / 2017-01-06

* feat(package): add wheel distribution

v0.2.1 / 2017-01-04

* fix(retrier): remove debug print statement

v0.2.0 / 2017-01-02

* feat(core): use seconds as default time unit (introduces API breaking changes)
* refactor(examples): update examples to use new time unit
* feat(contextmanager): adds context manager support
* feat(examples): add context manager example
* feat: add context managers support

v0.1.3 / 2016-12-30

* refactor(async_retrier): simplify coroutine wrapper
* feat(whitelist): add whitelist and blacklist support
* feat(tests): add missing test cases for whitelist
* feat(retry): pass error_evaluator param
* fix(retrier): cast delay to float
* fix(tests): use valid exception for Python 2.7
* feat(#6): add custom error whilelist and custom error evaluator function
* Merge pull request #8 from tsarpaul/master
* refactor(decorator): do not expose retrier instance

v0.1.2 / 2016-12-27

* fix(decorator): wrap retries instance per function call

v0.1.1 / 2016-12-27

* fix(`#2`_): handle and forward ``asyncio.CancelledError`` as non-retriable error

v0.1.0 / 2016-12-25

* First version

.. _#2:

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

riprova-0.2.5.tar.gz (19.3 kB view hashes)

Uploaded Source

Built Distribution

riprova-0.2.5-py2.py3-none-any.whl (25.4 kB view hashes)

Uploaded Python 2 Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page