How to capture API failure while using oauthlib.oauth2 fetch_token

Question

The Python3 fetch_token method in this library does not check the response status before consuming the response. If the API call it makes fails, then the response will be invalid and the script crashes. Is there something I can set so that an exception will be raised on a non-success response before the library can read the response?

import requests
from requests.auth import HTTPBasicAuth
from requests_oauthlib import OAuth2Session
from oauthlib.oauth2 import BackendApplicationClient
from oauthlib.oauth2 import OAuth2Error

AUTH_TOKEN_URL = "https://httpstat.us/500"  # For testing
AUTH = HTTPBasicAuth("anID", "aSecret")
CLIENT = BackendApplicationClient(client_id="anID")
SCOPES = "retailer.orders.write"
MAX_API_RETRIES = 4

class MyApp:
    def __init__(self):
        """Initialize ... and obtain initial auth token for request"""
        self.client = OAuth2Session(client=CLIENT)
        self.client.headers.update(
            {
                "Content-Type": "application/json"
            }
        )

        self.__authenticate()

    def __authenticate(self):
        """Obtain auth token."""
        server_errors = 0

        # This needs more work. fetch_token is not raising errors but failing
        # instead.
        while True:
            try:
                self.token = self.client.fetch_token(
                    token_url=AUTH_TOKEN_URL, auth=AUTH, scope=SCOPES
                )
                break

            except (OAuth2Error, requests.exceptions.RequestException) as e:
                server_errors = MyApp.__process_retry(
                    server_errors, e, None, MAX_API_RETRIES
                )

    @staticmethod
    def __process_retry(errors, exception, resp, max_retries):
        # Log and process retries
        # ...

        return errors + 1

MyApp()  # Try it out

Answer 1

You can add a "compliance hook" that will be passed the Response object from requests before the library attempts to parse it, like so:

def raise_on_error(response):
    response.raise_for_status()
    return response

self.client.register_compliance_hook('access_token_response', raise_on_error)

Depending on exactly when you may get errors, you might want to do this with 'refresh_token_response' and/or 'protected_request' as well. See the docstring for the register_compliance_hook method for more info.