types-requests-oauthlib

import os from typing import Dict, Any, Tuple from requests_oauthlib import OAuth2Session from requests.models import Response # --- Configuration (replace with your actual values or env vars) --- CLIENT_ID: str = os.environ.get('OAUTH_CLIENT_ID', 'your_client_id') CLIENT_SECRET: str = os.environ.get('OAUTH_CLIENT_SECRET', 'your_client_secret') AUTHORIZATION_BASE_URL: str = os.environ.get('OAUTH_AUTH_URL', 'https://example.com/oauth/authorize') TOKEN_URL: str = os.environ.get('OAUTH_TOKEN_URL', 'https://example.com/oauth/token') REDIRECT_URI: str = os.environ.get('OAUTH_REDIRECT_URI', 'https://example.com/callback') SCOPE: list[str] = ['read', 'write'] # Example scopes def run_oauth_flow() -> None: # 1. Create an OAuth2Session instance # The type hints from types-requests-oauthlib will check this. oauth: OAuth2Session = OAuth2Session(CLIENT_ID, redirect_uri=REDIRECT_URI, scope=SCOPE) # 2. Prepare the authorization request URL authorization_url: str state: str authorization_url, state = oauth.authorization_url(AUTHORIZATION_BASE_URL) print(f"Visit this URL to authorize: {authorization_url}") print(f"Remember state for CSRF protection: {state}") # --- Simulate authorization response --- # In a real application, the user would visit authorization_url, grant access, # and be redirected to REDIRECT_URI with a 'code' and 'state' in the URL. # For this example, we'll simulate the response URL. simulated_auth_response_url: str = f"{REDIRECT_URI}?code=AUTHORIZATION_CODE_FROM_PROVIDER&state={state}" print(f"\nSimulating callback URL: {simulated_auth_response_url}") # 3. Fetch the access token token: Dict[str, Any] = oauth.fetch_token( TOKEN_URL, client_secret=CLIENT_SECRET, authorization_response=simulated_auth_response_url ) print(f"\nAccess Token obtained: {token}") # 4. Use the obtained token to make a protected API request # The 'oauth' session automatically adds the authorization header. # The type hints for 'get' and 'Response' are provided by types-requests and types-requests-oauthlib. try: response: Response = oauth.get('https://example.com/api/protected_resource') response.raise_for_status() # Raise an exception for HTTP errors print(f"\nProtected resource content: {response.json()}") except Exception as e: print(f"\nError accessing protected resource: {e}") if __name__ == "__main__": run_oauth_flow()