The easiest way to retry operations
Project description
rtry
[
])(https://github.com/nikitanovosibirsk/rtry)
Installation
pip3 install rtry
Attempts
@retry(attempts=2)
def fn():
resp = requests.get("https://httpbin.org/status/500")
print(resp)
assert resp.status_code == 200
return resp
resp = fn()
# <Response [500]>
# <Response [500]>
# Traceback:
# AssertionError
Until
@retry(until=lambda r: r.status_code != 200, attempts=2)
def fn():
resp = requests.get("https://httpbin.org/status/500")
print(resp)
return resp
resp = fn()
# <Response [500]>
# <Response [500]>
Logger
Simple logger
@retry(until=lambda r: r.status_code != 200, attempts=2, logger=print)
def fn():
resp = requests.get("https://httpbin.org/status/500")
return resp
resp = fn()
# 1 <Response [500]> <function fn at 0x103dcd268>
# 2 <Response [500]> <function fn at 0x103dcd268>
Custom logger
def logger(attempt, result_or_exception, decorated):
logging.info("Attempt: %d, Result: %s", attempt, result_or_exception)
@retry(until=lambda r: r.status_code != 200, attempts=2, logger=logger)
def fn():
resp = requests.get("https://httpbin.org/status/500")
return resp
resp = fn()
# INFO:root:Attempt: 1, Result: <Response [500]>
# INFO:root:Attempt: 2, Result: <Response [500]>
Delay
Const delay
@retry(until=lambda r: r.status_code != 200, attempts=2, delay=0.1)
def fn():
resp = requests.get("https://httpbin.org/status/500")
return resp
started_at = time()
resp = fn()
ended_at = time()
print('Elapsed {:.2f}'.format(ended_at - started_at))
# Elapsed 2.11
Custom delay
from math import exp
@retry(until=lambda r: r.status_code != 200, attempts=2, delay=exp)
def fn():
resp = requests.get("https://httpbin.org/status/500")
return resp
started_at = time()
resp = fn()
ended_at = time()
print('Elapsed {:.2f}'.format(ended_at - started_at))
# Elapsed 11.79
Swallow
Fail on first exception
@retry(attempts=2, swallow=None, logger=print)
def fn():
resp = requests.get("http://127.0.0.1/status/500")
return resp
try:
resp = fn()
except Exception as e:
print(e)
# HTTPConnectionPool(host='127.0.0.1', port=80): Max retries exceeded with url: /status/500
Swallow only ConnectionError
from requests.exceptions import ConnectionError
@retry(attempts=2, swallow=ConnectionError, logger=print)
def fn():
resp = requests.get("http://127.0.0.1/status/500")
return resp
try:
resp = fn()
except Exception as e:
print(e)
# 1 HTTPConnectionPool(host='127.0.0.1', port=80): Max retries exceeded with url: /status/500
# 2 HTTPConnectionPool(host='127.0.0.1', port=80): Max retries exceeded with url: /status/500
# HTTPConnectionPool(host='127.0.0.1', port=80): Max retries exceeded with url: /status/500
Timeout
As argument
from retry import CancelledError
@retry(until=lambda r: r.status_code != 200, timeout=0.1)
def fn():
resp = requests.get("https://httpbin.org/status/500")
return resp
try:
resp = fn()
except CancelledError:
pass
As decorator
from retry import timeout, CancelledError
@timeout(0.1)
def fn():
resp = requests.get("https://httpbin.org/status/500")
return resp
try:
resp = fn()
except CancelledError:
pass
Project details
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.
Source Distribution
rtry-1.0.2.tar.gz
(5.2 kB
view hashes)
Built Distribution
rtry-1.0.2-py3-none-any.whl
(7.3 kB
view hashes)