Redo: Utilities to retry Python callables
Redo is a Python library that provides various mechanisms to add seamless retry functionality to any Python callable. It includes plain functions (redo.retry, redo.retry_async), decorators (redo.retriable, redo.retriable_async), and a context manager (redo.retrying). The current version is 3.0.0, released on July 17, 2024. It enables developers to integrate retry logic with customizable backoff strategies, making code more resilient to transient failures.
Warnings
- gotcha By default, `redo.retriable` and `redo.retry` catch `Exception` if `retry_exceptions` is not specified. This can mask underlying issues that should not be retried, making debugging difficult.
- gotcha Misconfiguring backoff parameters (`attempts`, `sleeptime`, `max_sleeptime`, `sleepscale`, `jitter`) can lead to excessive delays or a large number of retries, potentially overloading services or causing long wait times.
- gotcha Incorrectly mixing synchronous and asynchronous retry mechanisms (e.g., using `redo.retriable` with an `async def` function, or `redo.retriable_async` with a `def` function) will lead to runtime errors or unexpected behavior.
Install
-
pip install redo
Imports
- retry
from redo import retry
- retriable
from redo import retriable
- retrying
from redo import retrying
- retry_async
from redo import retry_async
- retriable_async
from redo import retriable_async
Quickstart
import random
import time
from redo import retriable
MAX_ATTEMPTS = 3
@retriable(
attempts=MAX_ATTEMPTS,
sleeptime=0.1, # initial sleep in seconds
max_sleeptime=1.0, # max sleep per retry
sleepscale=2, # exponential backoff factor
jitter=0.02, # random +/- seconds to add to sleeptime
retry_exceptions=(RuntimeError, ConnectionError) # specify exceptions to retry
)
def flaky_operation(attempt_num: int):
# Simulate a network call or I/O operation that might fail
if random.random() < 0.7 and attempt_num < MAX_ATTEMPTS: # 70% chance of failure for initial attempts
print(f"Attempt {attempt_num}: Operation failed, retrying...")
raise ConnectionError("Failed to connect to service")
print(f"Attempt {attempt_num}: Operation successful!")
return "Success"
if __name__ == "__main__":
print("Starting flaky operation...")
try:
result = flaky_operation(attempt_num=1)
print(f"Final result: {result}")
except Exception as e:
print(f"Operation ultimately failed after multiple retries: {e}")