Recursive Error Handling in Python with a limited amount of tries

Question

I have a class method that is set up to make requests to an API and I'm having issues handling the possible errors that could arise by trying to communicate with that server. The idea is if an error were to pop up, then it would call on itself until either it no longer needs to (because it worked), or it reaches a limit I set up for it; in this case, I would want it to just simply raise an exception that would be caught somewhere else in the code. What I don't know is how to properly return the response if it crapped out the first time it made the call and had to call on itself from the except block.

I'm not 100% clear if this actually gets the job done. Additionally, I dont know if this is the best way to do it. I know a possible suggestion is to make error handling more specific, given the requests library myriad exceptions, but I figured that any exception should just be handled by trying X amount of time and then quitting if it doesn't work.

def getRequest(self, endpoint, attempts):
        baseUrl = self.baseUrl
        url = baseUrl + endpoint
        if self.expirationT <= datetime.datetime.now():
            self.token, self.expirationT = self.auth(client_id, client_secret)
        else:
            pass
        try:
            response = requests.get(url, auth = BearerAuth(self.token))
            response.raise_for_status()
        except:
            if attempts < 20:
                time.sleep(3)
                response = self.getRequest(endpoint, attempts + 1)
                return response
            else:
                raise Exception
        else:
            return response

Answer 1

I know a possible suggestion is to make error handling more specific, given the requests library myriad exceptions, but I figured that any exception should just be handled by trying X amount of time and then quitting if it doesn't work.

I disagree here. Not just any exception should be handled. It's possible to have typo'd a bug into your code inside the try, and you definitely don't want try masking a bug.

If you check the source (or documentation), you'll see that the requests exceptions all seem to inherit from RequestException. If you really want to handle every possible request exception the same, I would catch the base class RequestException instead. I still don't think this is a good idea though without doing any logging. There may very well be a RequestException that gets thrown at some point which indicates that you accidentally gave the request bad data, not that there was a problem with the request being carried out using good data. I'd check the docs for the methods used and figure out what exact exceptions you want to retry on.

---

This also doesn't need to be recursion. In this case, nothing bad will likely happen because you have a limit of 20 retries, which isn't enough to exhaust the stack any sane case. If you ever increase that limit up to 1000 though, you may run into real problems.

I think this could be done pretty easily using a while True loop. The first two lines of the function seem to be essentially constants, so they don't need to be recomputed every time. Everything under those lines though can be stuck in a loop.

def getRequest(self, endpoint, max_attempts=20, retry_delay=3):
    baseUrl = self.baseUrl
    url = baseUrl + endpoint

    attempts = 0
    while True:
        if self.expirationT <= datetime.datetime.now():
            self.token, self.expirationT = self.auth(client_id, client_secret)

        try:
            response = requests.get(url, auth=BearerAuth(self.token))
            response.raise_for_status()
            return response

        except requests.RequestException as e:
            attempts += 1

            if attempts < max_attempts:
                time.sleep(retry_delay)

            else:
                raise RuntimeError("Max number of retires met.")

                # Or to preserve in the trace the original problem that caused the error:
                # raise RuntimeError("Max number of retires met.") from e

Things to note:

• To retry now, instead of manually recursing, I'm just letting control fall out of the except so that the while can restart again.

• Instead of attempts being a parameter, I just made it a local variable which is incremented inside of the except.

• I'm throwing a more specialized exception with an informative error message. Throwing the generic Exception makes life more difficult for the users of your code. Ideally, they should be able to pick and choose what exceptions they handle and when. Throwing Exception though forces them to catch your errors. RuntimeError isn't really the best exception here, but I couldn't think of a good built-in one for this purpose. You may want to make a custom exception for this case:

  class TooManyRetries(Exception):
      pass
  
  . . .
  
  raise TooManyRetries("Max number of retires met.")
  

• I got rid of the else: pass. That isn't necessary.

• You had two "magic numbers": 20 and 3 to mean the max number of attempts and the retry delay. I don't think it's a good idea to have those hard coded though. What if you want to change either at some point? You'd have to edit the code. I made them parameters of the function, defaulting to the values that you had. If you don't specify them, the behavior will be as you had before, but now they can be easily changed as needed.

Answer 2

I agree with the other answer by @Carcigenicate. Putting this into a loop instead of using a recursive design makes it a lot easier to understand. However, the explicit while loop can be simplified even further by using a for loop instead. This makes it even more readable, IMO.

def get_request(self, end_point, max_attempts=20, retry_delay=3):
    """Get the page at `self.base_url + end_point`.
    If the request fails due to a request error, retry up to `max_attempts` times,
    with a delay of `retry_delay` seconds.
    """
    url = self.base_url + endpoint
    for _ in range(max_attempts):
        if self.expiration_time <= datetime.datetime.now():
            self.token, self.expiration_time = self.auth(client_id, client_secret)
        try:
            response = requests.get(url, auth=BearerAuth(self.token))
            response.raise_for_status()
            return response
        except requests.RequestException:
            time.sleep(retry_delay)
    raise RuntimeError(f"Maximum number of retries ({max_attempts}) reached.")

In addition, Python has an official style-guide, PEP8. It recommends using lower_case for functions and variables and PascalCase only for classes.

You should also add a docstring to document your functions, as I have done for this method.

You might also want to add some logging to this, so you can debug it if necessary:

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("Requests")

...

def get_request(self, end_point, max_attempts=20, retry_delay=3):
    """Get the page at `self.base_url + end_point`.
    If the request fails due to a request error, retry up to `max_attempts` times,
    with a delay of `retry_delay` seconds.
    """
    url = self.base_url + endpoint
    for attempt in range(max_attempts):
        if self.expiration_time <= datetime.datetime.now():
            self.token, self.expiration_time = self.auth(client_id, client_secret)
        try:
            response = requests.get(url, auth=BearerAuth(self.token))
            response.raise_for_status()
            return response
        except requests.RequestException:
            logger.exception("Attempt %s out of %s failed for URL %s",
                             attempt + 1, max_attempts, url)
            time.sleep(retry_delay)
    raise RuntimeError(f"Maximum number of retries ({max_attempts}) reached.")

Answer 3

The entire question essentially spins around and calls for a good "Retry/MaxRetry nextwork requests" strategy.

And just for this we have a flexible and reach Retry configuration - urllib3.util.Retry, which in combination with well-known requests lib provides granular control over the conditions under which we retry a request.

Among a bunch of options it provides:

class urllib3.util.Retry(total=10, connect=None, read=None, redirect=None, status=None, method_whitelist=frozenset(['HEAD', 'TRACE', 'GET', 'PUT', 'OPTIONS', 'DELETE']), status_forcelist=None, backoff_factor=0, raise_on_redirect=True, raise_on_status=True, history=None, respect_retry_after_header=True, remove_headers_on_redirect=frozenset(['Authorization']))

I'll mention just some crucial ones:

• total. Total number of retries to allow.
• status_forcelist(iterable). A set of integer HTTP status codes that we should force a retry on.
  In my sample scheme below I used all error HTTP codes to be considered here. But you can restrict that iterable by passing the argument with selected codes like status_forcelist=(500, 502, 503, 504).

• backoff_factor. A backoff factor to apply between attempts after the second try (most errors are resolved immediately by a second try without a delay). urllib3 will sleep for:

  {backoff factor} * (2 ** ({number of total retries} - 1))
  

  seconds. If the backoff_factor is 0.1, then sleep() will sleep for [0.0s, 0.2s, 0.4s, …] between retries. It will never be longer than Retry.BACKOFF_MAX.
  This would be a more "thought-out" alternative to a constant delay.

---

To establish the above Retry/MaxRetry strategy we run through the following steps:

• creating instance of Retry component with the needed options
• creating requests.adapters.HTTPAdapter adapter instance with passing in created Retry component as its max_retries parameter: HTTPAdapter(max_retries=self._retry)
• creating requests session with self.session = requests.Session() and mounting it to a base url (self.session.mount(self.base_url, self._adapter)). You can mount session to multiple base urls/prefixes.

Since you posted an instance method def getRequest(self, ... I assume that that's the context of some API client/wrapper. Below is a generic sample scheme of using the described strategy. So you can easily adjust/extend your API client class appropriately (and get rid of loops or recursion in "Retry" intention).

import requests
from requests.adapters import HTTPAdapter
from requests.exceptions import RetryError
from urllib3.exceptions import MaxRetryError
from urllib3.util.retry import Retry
import datetime


# .... your constants/variables

class MyAPIClient:
    ERROR_CODES = tuple(code for code in requests.status_codes._codes if code >= 400)

    def __init__(self, base_url, max_retries=5, backoff_factor=0.2):
        self.base_url = base_url
        self._max_retries = max_retries
        self._retry = Retry(total=max_retries,
                           backoff_factor=backoff_factor,
                           status_forcelist=MyAPIClient.ERROR_CODES)
        self._adapter = HTTPAdapter(max_retries=self._retry)

        self.session = requests.Session()
        self.session.mount(self.base_url, self._adapter)

    def get_request(self, endpoint):
        url = self.base_url + endpoint

        if self.expirationT <= datetime.datetime.now():
            self.token, self.expirationT = self.auth(client_id, client_secret)

        try:
            response = self.session.get(url, auth=BearerAuth(self.token))
        except (MaxRetryError, RetryError) as ex:
            # optional actions/logging here. Otherwise - try/except can be just eliminated
            raise
        return response