A generic retry package for Python.
MIT License
A generic retry package for Python.
pip install retry_plus
For detailed usage and examples, refer to the Conditions Documentation.
from retry import Retry, stop_after_attempt, wait_exponential
import time
@Retry(
stop_condition=stop_after_attempt(5),
wait_condition=wait_exponential(multiplier=1, min_wait=1, max_wait=10),
retry_on_exceptions=(ValueError,),
retry_on_result=lambda result: result != "Success"
)
def unreliable_sync_function():
print("Trying to perform an unreliable operation.")
if time.time() % 2 < 1:
raise ValueError("Simulated transient error.")
return "Success"
try:
result = unreliable_sync_function()
print(f"Function succeeded with result: {result}")
except Exception as e:
print(f"Function failed after retries with exception: {e}")
import asyncio
from retry import Retry, stop_after_attempt, wait_exponential
import time
@Retry(
stop_condition=stop_after_attempt(5),
wait_condition=wait_exponential(multiplier=1, min_wait=1, max_wait=10),
retry_on_exceptions=(ValueError,),
retry_on_result=lambda result: result != "Success"
)
async def unreliable_async_function():
print("Trying to perform an unreliable operation.")
if time.time() % 2 < 1:
raise ValueError("Simulated transient error.")
return "Success"
async def main():
try:
result = await unreliable_async_function()
print(f"Function succeeded with result: {result}")
except Exception as e:
print(f"Function failed after retries with exception: {e}")
asyncio.run(main())
from retry import Retry, stop_after_attempt, wait_exponential
import time
try:
with Retry(
stop_condition=stop_after_attempt(3),
wait_condition=wait_exponential(multiplier=1, min_wait=1, max_wait=5),
retry_on_exceptions=(ValueError,)
):
print("Trying block operation.")
if time.time() % 2 < 1:
raise ValueError("Simulated transient error.")
print("Block operation succeeded.")
except Exception as e:
print(f"Context manager failed after retries with exception: {e}")
from retry import Retry, stop_after_delay, wait_random_exponential
import time
@Retry(
stop_condition=stop_after_delay(20), # Stop after 20 seconds
wait_condition=wait_random_exponential(multiplier=1, max_seconds=10), # Exponential backoff with randomness
retry_on_exceptions=(ValueError,)
)
def unreliable_function():
print("Trying to perform an unreliable operation.")
if time.time() % 2 < 1:
raise ValueError("Simulated transient error.")
return "Success"
try:
result = unreliable_function()
print(f"Function succeeded with result: {result}")
except Exception as e:
print(f"Function failed after retries with exception: {e}")
import requests
from retry import Retry, stop_after_attempt, wait_exponential
# Define a function to make an HTTP request
@Retry(
stop_condition=stop_after_attempt(5), # Retry up to 5 times
wait_condition=wait_exponential(multiplier=1, min_wait=1, max_wait=10), # Exponential backoff
retry_on_exceptions=(requests.RequestException,), # Retry on any requests exception
retry_on_result=lambda result: result.status_code != 200 # Retry if the status code is not 200
)
def fetch_data_from_api(url):
print(f"Trying to fetch data from {url}")
response = requests.get(url)
response.raise_for_status() # Raise an HTTPError on bad status
return response
# Use the function with a simulated unreliable endpoint
try:
# Simulating a service that returns 500 Internal Server Error 50% of the time
data = fetch_data_from_api("https://httpbin.org/status/500")
print("Data fetched successfully:", data)
except Exception as e:
print(f"Failed to fetch data after retries. Error: {e}")
# Use the function with a simulated successful endpoint
try:
# Simulating a service that returns 200 OK
data = fetch_data_from_api("https://httpbin.org/status/200")
print("Data fetched successfully:", data)
except Exception as e:
print(f"Failed to fetch data after retries. Error: {e}")
If stop_condition
and wait_condition
are not provided, the following defaults will be used:
from retry import Retry
import time
@Retry()
def default_unreliable_function():
print("Trying to perform an unreliable operation.")
if time.time() % 2 < 1:
raise ValueError("Simulated transient error.")
return "Success"
try:
result = default_unreliable_function()
print(f"Function succeeded with result: {result}")
except Exception as e:
print(f"Function failed after retries with exception: {e}")
To run the tests, use pytest
:
pytest
We welcome contributions to improve the retry package. Here are some ways you can contribute:
git clone https://github.com/talaatmagdyx/retry_plus.git
git checkout -b my-new-feature
git commit -am 'Add some feature'
git push origin my-new-feature
All notable changes to this project will be documented in this section.