Small and versatile library to retry failed operations using different backoff strategies
Project description
riprova 
Small, generic and versatile Python library providing retry mechanisms with multiple backoff strategies for failed operations.
For an brief introduction about backoff mechanisms for potential failed operations, read this article.
Features
Retry decorator for simple and idiomatic consumption.
Maximum retry timeout support.
Automatically retry operations on raised exceptions.
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 back-off strategies: constant, fibonacci and exponential back-offs.
Pluggable, custom back-off strategies.
Supports asynchronous coroutines with async/await and yield from syntax.
Lightweight small library with zero embedding cost.
Works with Python +2.6, 3.0+ and PyPy.
Backoff strategies
List of built-in backoff strategies.
You can also implement your own one easily. See ConstantBackoff for an implementation reference.
Installation
Using pip package manager (requires pip 1.8+. Upgrade it running: pip install -U pip):
pip install -U riprovaOr install the latest sources from Github:
pip install -e git+git://github.com/h2non/riprova.git#egg=riprovaAPI
Examples
You can see all the featured examples from the documentation.
Basic usage examples:
import riprova
@riprova.retry
def task():
"""Retry operation if it fails with constant backoff"""
@riprova.retry(backoff=riprova.ExponentialBackOff(factor=0.5))
def task():
"""Retry operation if it fails using exponential backoff"""
@riprova.retry(timeout=10000)
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')
# 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
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')
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 request
fetch('http://server.com')License
MIT - Tomas Aparicio
History
0.1.0 (2016-12-25)
First version
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
