如何在使用oauthlib.oauth2的fetch_token时捕获API失败并提前抛异常

Great question! I’ve run into this exact issue before—fetch_token has a quirk where it doesn’t check HTTP status codes before trying to parse the response body. When you get a non-2xx (like that 500 error in your test), it tries to treat the error page as a valid OAuth2 token response, which blows up instead of triggering a clean exception. But don’t worry, there’s an easy fix using requests response hooks that will make non-success responses throw exceptions before the library tries to parse them.

The Solution: Add a Response Hook to Check Status Codes #

Since OAuth2Session is built on top of requests.Session, we can add a hook that runs immediately after receiving a response (before oauthlib parses it) to validate the status code. Here’s how to modify your code:

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

def check_response_status(response, *args, **kwargs):
    # Raise an exception for any non-2xx HTTP status code
    response.raise_for_status()

class MyApp:
    def __init__(self):
        """Initialize ... and obtain initial auth token for request"""
        self.client = OAuth2Session(client=CLIENT)
        # Add the status-checking hook to the session
        self.client.hooks["response"].append(check_response_status)
        self.client.headers.update(
            {
                "Content-Type": "application/json"
            }
        )
        self.__authenticate()

    def __authenticate(self):
        """Obtain auth token."""
        server_errors = 0
        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 (example implementation)
        print(f"Retry {errors + 1}/{max_retries} failed: {str(exception)}")
        if errors >= max_retries - 1:  # Stop after max attempts
            print("Max retries reached, aborting.")
            raise exception
        return errors + 1

MyApp() # Try it out

How This Works #

1. The Hook Function: check_response_status calls response.raise_for_status(), which throws a requests.exceptions.HTTPError for any response with a status code ≥400.
2. Hook Integration: We add this function to the session’s response hooks. Hooks execute right after the response is received, before oauthlib tries to parse the body as an OAuth2 token.
3. Exception Handling: Your existing except block already catches requests.exceptions.RequestException (which HTTPError inherits from), so your retry logic will kick in cleanly instead of the script crashing from a failed JSON parse.

Alternative: Manual Request Handling (For Full Control) #

If you want even more control over the token request flow, you can manually send the request, check the status code yourself, then pass the valid response to oauthlib for parsing. Here’s a quick example:

def __authenticate(self):
    server_errors = 0
    while True:
        try:
            # Manually send the client credentials request
            response = requests.post(
                AUTH_TOKEN_URL,
                auth=AUTH,
                data={"grant_type": "client_credentials", "scope": SCOPES}
            )
            response.raise_for_status()  # Validate status first
            # Let oauthlib parse the valid response body
            self.token = CLIENT.parse_request_body_response(response.text)
            self.client.token = self.token
            break
        except (OAuth2Error, requests.exceptions.RequestException) as e:
            server_errors = MyApp.__process_retry(
                server_errors, e, response, MAX_API_RETRIES
            )

This approach avoids fetch_token entirely, giving you full control over every step, but the hook method is simpler if you want to stick with the OAuth2Session API you’re already using.

内容的提问来源于stack exchange，提问作者George Shaw