18

I have multiple functions in my script which does a REST API api requests.As i need to handle the error scenarios i have put a retry mechanism as below.

no_of_retries = 3
def check_status():
    for i in range(0,no_of_retries):
        url = "http://something/something"
        try:
            result = requests.get(url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()
            if 'error' not in result:
                return result
            else:
                continue
        except Exception as e:
            continue
    return None

I have several different methods which does similar operation. How can we do it better way to avoid duplication may be using decorators.

MrLeeh
  • 4,895
  • 3
  • 29
  • 46
PGS
  • 906
  • 2
  • 13
  • 34

6 Answers6

17

if you do not mind installing a library you could use the tenacity (github.com/jd/tenacity) module. one of their examples:

import random
from tenacity import retry, stop_after_attempt

# @retry  # retry forever
@retry(stop=stop_after_attempt(3))
def do_something_unreliable():
    if random.randint(0, 10) > 1:
        raise IOError("Broken sauce, everything is hosed!!!111one")
    else:
        return "Awesome sauce!"

print(do_something_unreliable())

this also allows you to specify the number of tries or seconds you want to keep retrying.

for your case this might look something like this (not tested!):

@retry(stop=stop_after_attempt(3))
def retry_get():
    result = requests.get(
            url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()
    if 'error' not in result:
        raise RequestException(result)
hiro protagonist
  • 40,708
  • 13
  • 78
  • 98
15

You can use a decorator like this and handle your own exception.

def retry(times, exceptions):
    """
    Retry Decorator
    Retries the wrapped function/method `times` times if the exceptions listed
    in ``exceptions`` are thrown
    :param times: The number of times to repeat the wrapped function/method
    :type times: Int
    :param Exceptions: Lists of exceptions that trigger a retry attempt
    :type Exceptions: Tuple of Exceptions
    """
    def decorator(func):
        def newfn(*args, **kwargs):
            attempt = 0
            while attempt < times:
                try:
                    return func(*args, **kwargs)
                except exceptions:
                    print(
                        'Exception thrown when attempting to run %s, attempt '
                        '%d of %d' % (func, attempt, times)
                    )
                    attempt += 1
            return func(*args, **kwargs)
        return newfn
    return decorator

@retry(times=3, exceptions=(ValueError, TypeError))
def foo1():
    print('Some code here ....')
    print('Oh no, we have exception')
    raise ValueError('Some error')

foo1()
mrkiril
  • 601
  • 7
  • 10
  • 4
    I started writing my own and then realized it must have been done a million times before. This was perfect except I added a timeout param with a time.sleep() when attempt=times before returning the func(). Thanks! – Sean DiZazzo Jul 31 '21 at 16:34
3

The third-party retry module is now widely accepted for this. You can also pass the list of exceptions to retry for, number of retries, delays, maximum delay, exponential back-off, etc.

$ pip install retry

Example usage:

from retry import retry

@retry(ZeroDivisionError, tries=3, delay=2)
def make_trouble():
    '''Retry on ZeroDivisionError, raise error after 3 attempts, sleep 2 seconds between attempts.'''
MohitC
  • 3,952
  • 1
  • 29
  • 49
1

Instead of using decorators, the probably better solution is to move the request to its own function, arriving at a structure similar to this:

no_of_retries = 3

def make_request(url):
    for i in range(0,no_of_retries):
        try:
            result = requests.get(url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()
            if 'error' not in result:
                return result
            else:
                continue
        except Exception as e:
            continue
    return result

def check_status():
    result = make_request("http://something/status")

def load_file():
    result = make_request("http://something/file")

This way, you avoid duplicate code while encapsulating the request. If you were to use a decorator, you would need to wrap the whole load_file() method which would prevent you from further processing the request's result within this function.

IonicSolutions
  • 2,442
  • 1
  • 15
  • 30
1

Production level example

import logging
import time
import functools
import traceback

LOG_FORMAT = "%(asctime)s - %(levelname)s - %(pathname)s - %(funcName)s - %(lineno)d -msg: %(message)s"
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)


def retry(retry_num, retry_sleep_sec):
    """
    retry help decorator.
    :param retry_num: the retry num; retry sleep sec
    :return: decorator
    """
    def decorator(func):
        """decorator"""
        # preserve information about the original function, or the func name will be "wrapper" not "func"
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            """wrapper"""
            for attempt in range(retry_num):
                try:
                    return func(*args, **kwargs)  # should return the raw function's return value
                except Exception as err:   # pylint: disable=broad-except
                    logging.error(err)
                    logging.error(traceback.format_exc())
                    time.sleep(retry_sleep_sec)
                logging.error("Trying attempt %s of %s.", attempt + 1, retry_num)
            logging.error("func %s retry failed", func)
            raise Exception('Exceed max retry num: {} failed'.format(retry_num))

        return wrapper

    return decorator

usage

# this means try your function 5 times, each time sleep 60 seconds
@retry(5, 60)
def your_func():
    pass

Formal reference: https://peps.python.org/pep-0318/

wahaha
  • 41
  • 3
  • learn more about decorators: [realpython](https://realpython.com/primer-on-python-decorators/) – wahaha May 20 '22 at 08:54
0

Using functools on top of mrkiril's answer:

from functools import wraps, partial

def retry(f=None, times=10):
    if f is None:
        return partial(retry, times=times)

    @wraps(f)
    def wrap(*args, **kwargs):
        attempt = 0
        while attempt < times:
            try:
                return f(*args, **kwargs)
            except:
                print(f"{f.__name__}, attempt {attempt} of {times}")
                attempt += 1
        return f(*args, **kwargs)
    return wrap

Then, wrap your function like the following:

import random

@retry
def foo():
    if random.randint(0, 5) != 0:
        raise Exception
mataney
  • 116
  • 3