Skip to main content

The easiest way to retry operations

Project description

rtry

License Drone Codecov PyPI Python Version

Installation

pip3 install rtry

Documentation


Timeout

As context manager
from rtry import timeout, CancelledError

try:
    with timeout(3.0):
        resp = requests.get("https://httpbin.org/status/200")
except CancelledError:
    raise
else:
    print(resp)
As context manager (silent)
from rtry import timeout, CancelledError

resp = None
with timeout(3.0, exception=None):
    resp = requests.get("https://httpbin.org/status/200")
As context manager (asyncio)
import asyncio
import aiohttp
from rtry import timeout, CancelledError

async def main():
    try:
        async with aiohttp.ClientSession() as session, timeout(3.0):
            async with session.get("https://httpbin.org/status/200") as resp:
                return resp
    except CancelledError:
        raise
    else:
        print(resp)

asyncio.run(main())
As decorator
from rtry import timeout, CancelledError

@timeout(3.0)
def fn():
    resp = requests.get("https://httpbin.org/status/200")
    return resp

try:
    resp = fn()
except CancelledError:
    raise
else:
    print(resp)
As decorator (asyncio)
import asyncio
import aiohttp
from rtry import timeout, CancelledError

@timeout(3.0)
async def fn():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://httpbin.org/status/200") as resp:
            return resp

async def main():
    try:
        resp = await fn()
    except CancelledError:
        raise
    else:
        print(resp)

asyncio.run(main())
As argument
from rtry import retry, CancelledError

@retry(until=lambda r: r.status_code != 200, timeout=3.0)
def fn():
    resp = requests.get("https://httpbin.org/status/200")
    return resp

try:
    resp = fn()
except CancelledError:
    raise
else:
    print(resp)

Retry

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.monotonic()
resp = fn()
ended_at = time.monotonic()
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.monotonic()
resp = fn()
ended_at = time.monotonic()
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

AsyncIO

import asyncio
import aiohttp
from rtry import retry

@retry(attempts=2)
async def fn():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://httpbin.org/status/500") as resp:
            print(resp)
            assert resp.status == 200
            return resp

async def main():
    resp = await fn()
    # <ClientResponse(https://httpbin.org/status/500) [500 INTERNAL SERVER ERROR]>
    # <ClientResponse(https://httpbin.org/status/500) [500 INTERNAL SERVER ERROR]>
    # Traceback
    #   AssertionError

asyncio.run(main())

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for rtry, version 1.2.0
Filename, size File type Python version Upload date Hashes
Filename, size rtry-1.2.0-py3-none-any.whl (8.3 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size rtry-1.2.0.tar.gz (6.3 kB) File type Source Python version None Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page