riprova 0.2.1

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

riprova

riprova (meaning retry in Italian) is a small, general-purpose and versatile Python library providing retry mechanisms with multiple backoff strategies for failed operations.

It’s domain agnostic and can be used within any code base.

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

Features

• 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.

• Custom retry evaluator function, useful to determine when an operation failed or not.

• 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 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.

Installation

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

pip install -U riprova

Or install the latest sources from Github:

pip install -e git+git://github.com/h2non/riprova.git#egg=riprova

API

• 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

Examples

You can see more featured examples from the documentation site.

Basic usage examples:

import riprova

@riprova.retry
def task():
    """Retry operation if it fails with constant backoff (default)"""

@riprova.retry(backoff=riprova.ConstantBackoff(retries=5))
def task():
    """Retry operation if it fails with custom max number of retry attempts"""

@riprova.retry(backoff=riprova.ExponentialBackOff(factor=0.5))
def task():
    """Retry operation if it fails using exponential backoff"""

@riprova.retry(timeout=10)
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))

@riprova.retry(on_retry=on_retry)
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

@riprova.retry(evaluator=evaluator)
def task():
    """Use a custom evaluator function to determine if the operation failed or not"""

@riprova.retry
async def task():
    """Asynchronous coroutines are also supported :)"""

Retry failed HTTP requests:

import pook
import requests
from riprova import retry

# Define HTTP mocks to simulate failed requests
pook.get('server.com').times(3).reply(503)
pook.get('server.com').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
fetch('http://server.com')

License

MIT - Tomas Aparicio

History

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