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/awaitandyield fromsyntax. - 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.
List of built-in backoff strategies.
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):
pip install -U riprovaOr install the latest sources from Github:
pip install -e git+git://github.com/h2non/riprova.git#egg=riprova- 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:
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')MIT - Tomas Aparicio
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent