使用requests与requests_oauthlib对接OAuth私有Web服务求助

Hey there! Let's break down how to get this OAuth password grant flow working for your private web service using Python. I'll walk you through the code step by step, with explanations for the key bits you mentioned.

Step 1: Install Required Libraries #

First, make sure you have both requests and requests_oauthlib installed. If not, run this command in your terminal:

pip install requests requests_oauthlib

Step 2: Full Implementation Code #

Here's a complete example that handles both fetching the OAuth token and using it to access your data endpoint:

import requests
from requests_oauthlib import OAuth2Session

# Your provided credentials and URLs
AUTH_URL = 'https://foobar.url.com/oauth/token'
DATA_URL = 'https://foobar.url.com/rest/v1/aname/overview'
API_KEY = 'Basic Y2xpY2tleS1hcHA6YXBwLWFjY2Vzcw=='
USERNAME = 'ausername'
PASSWORD = 'apassword'
GRANT_TYPE = 'password'

def get_oauth_token():
    # Set up headers with your API Key as required
    headers = {
        'Authorization': API_KEY,
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    
    # Payload for the token request (password grant flow)
    payload = {
        'grant_type': GRANT_TYPE,
        'username': USERNAME,
        'password': PASSWORD
    }
    
    try:
        # Send POST request to retrieve the access token
        response = requests.post(AUTH_URL, headers=headers, data=payload)
        response.raise_for_status()  # Trigger error for HTTP status codes >=400
        token_data = response.json()
        return token_data.get('access_token')
    except requests.exceptions.RequestException as e:
        print(f"Error fetching OAuth token: {e}")
        return None

def fetch_data(access_token):
    if not access_token:
        print("No valid access token provided.")
        return None
    
    # Use the access token to authenticate the data request
    headers = {
        'Authorization': f'Bearer {access_token}'
    }
    
    try:
        # Send GET request to the protected data endpoint
        response = requests.get(DATA_URL, headers=headers)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error fetching data: {e}")
        return None

# Run the workflow
if __name__ == '__main__':
    access_token = get_oauth_token()
    if access_token:
        data = fetch_data(access_token)
        if data:
            print("Successfully fetched data:")
            print(data)

Key Details Explained #

• API Key Placement: We're adding your API_KEY directly to the Authorization header of the token request, exactly as you specified. This authenticates your client application to the OAuth server before requesting user-level access.
• Password Grant Flow: The payload uses grant_type=password alongside the user's credentials—this tells the OAuth server you're using the password flow, which is common for trusted internal applications.
• Access Token Usage: Once we receive the access_token, we attach it to the Authorization header (prefixed with Bearer) to make authenticated requests to the protected data endpoint.
• Error Handling: Basic error handling is included to catch HTTP errors and network issues, making it easier to debug problems like invalid credentials or server downtime.

Quick Notes for Production #

• Never hardcode credentials like API_KEY, username, or password in your code. Use environment variables or a secure secrets manager instead.
• Some OAuth servers might require extra parameters (like scope or client_id). If you run into issues, check the service's official OAuth documentation for any missing required fields.

内容的提问来源于stack exchange，提问作者Hans van Schaick