OAuth2 Flow Examples

This page demonstrates various OAuth2 authentication flows using HTTP Client.

Client Credentials Flow

Basic Client Credentials

from requestforge import (
    HttpClient,
    HttpClientConfigBuilder,
    TokenManager,
    ClientCredentialsTokenProvider,
    InMemoryTokenStorage
)

# Create OAuth2 provider
provider = ClientCredentialsTokenProvider(
    token_url='https://auth.example.com/oauth/token',
    client_id='your-client-id',
    client_secret='your-client-secret',
    service_name='example-api',
    scope='read write'
)

# Create token manager
token_manager = TokenManager(
    provider=provider,
    storage=InMemoryTokenStorage()
)

# Configure HTTP client
config = (
    HttpClientConfigBuilder()
    .with_base_url('https://api.example.com')
    .with_token_auth(token_manager=token_manager)
    .build()
)

client = HttpClient(config)

# Token automatically fetched, cached, and injected
response = client.get('/users')

# Token automatically refreshed when expired
response = client.get('/posts')

With Django Cache

from requestforge import (
    TokenManager,
    ClientCredentialsTokenProvider,
    DjangoCacheTokenStorage
)

# Provider
provider = ClientCredentialsTokenProvider(
    token_url='https://auth.example.com/oauth/token',
    client_id='your-client-id',
    client_secret='your-client-secret',
    service_name='example-api'
)

# Django cache storage (shared across processes)
storage = DjangoCacheTokenStorage(
    cache_alias='default',
    key_prefix='oauth_tokens'
)

# Token manager
token_manager = TokenManager(provider, storage)

# Django settings.py
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
    }
}

With Header Authentication

from requestforge import ClientCredentialsTokenProvider

# Send credentials in Authorization header (HTTP Basic)
provider = ClientCredentialsTokenProvider(
    token_url='https://auth.example.com/oauth/token',
    client_id='your-client-id',
    client_secret='your-client-secret',
    service_name='example-api',
    auth_method='header'  # Use HTTP Basic auth
)

With Extra Parameters

from requestforge import ClientCredentialsTokenProvider

provider = ClientCredentialsTokenProvider(
    token_url='https://auth.example.com/oauth/token',
    client_id='your-client-id',
    client_secret='your-client-secret',
    service_name='example-api',
    scope='read write',
    extra_params={
        'audience': 'https://api.example.com',
        'grant_type': 'client_credentials'
    }
)

Password Grant Flow

User Authentication

from requestforge import (
    TokenManager,
    PasswordGrantTokenProvider,
    InMemoryTokenStorage
)

# Password grant provider
provider = PasswordGrantTokenProvider(
    token_url='https://auth.example.com/oauth/token',
    client_id='your-client-id',
    client_secret='your-client-secret',
    username='user@example.com',
    password='user-password',
    service_name='example-api',
    scope='profile email'
)

# Token manager
token_manager = TokenManager(
    provider=provider,
    storage=InMemoryTokenStorage()
)

# Configure client
config = (
    HttpClientConfigBuilder()
    .with_base_url('https://api.example.com')
    .with_token_auth(token_manager=token_manager)
    .build()
)

client = HttpClient(config)
response = client.get('/user/profile')

Token Refresh

Automatic Refresh

from requestforge import (
    TokenManager,
    ClientCredentialsTokenProvider
)

provider = ClientCredentialsTokenProvider(
    token_url='https://auth.example.com/oauth/token',
    client_id='your-client-id',
    client_secret='your-client-secret',
    service_name='example-api'
)

token_manager = TokenManager(provider)

# First call: fetches token
token = token_manager.get_token()
print(f"Token: {token.access_token}")
print(f"Expires: {token.expires_at}")

# Subsequent calls: returns cached token
token = token_manager.get_token()

# After expiration: automatically refreshes
token = token_manager.get_token()

Manual Refresh

from requestforge import TokenManager

token_manager = TokenManager(provider)

# Force refresh token
new_token = token_manager.force_refresh()

# Invalidate token
token_manager.invalidate_token()

# Next call will fetch new token
token = token_manager.get_token()

Auth Retry on 401

Automatic Token Refresh on 401

from requestforge import (
    HttpClientConfigBuilder,
    HttpClient,
    TokenManager,
    SimpleAuthRetryStrategy
)

# Token manager
token_manager = TokenManager(provider)

# Auth retry strategy
auth_retry = SimpleAuthRetryStrategy(
    max_retries=1,  # Retry once on 401
    delay=0.5       # Wait 500ms before retry
)

# Configure client with auth retry
config = (
    HttpClientConfigBuilder()
    .with_base_url('https://api.example.com')
    .with_token_auth(
        token_manager=token_manager,
        auth_retry_strategy=auth_retry
    )
    .build()
)

client = HttpClient(config)

# If 401 error:
# 1. Token manager refreshes token
# 2. Request retried with new token
response = client.get('/protected-resource')

Exponential Auth Retry

from requestforge import ExponentialAuthRetryStrategy

# Exponential backoff for auth retries
auth_retry = ExponentialAuthRetryStrategy(
    max_retries=2,
    base_delay=0.5,
    max_delay=5.0,
    multiplier=2.0,
    jitter=True
)

config = (
    HttpClientConfigBuilder()
    .with_base_url('https://api.example.com')
    .with_token_auth(
        token_manager=token_manager,
        auth_retry_strategy=auth_retry
    )
    .build()
)

Custom OAuth2 Provider

Custom Token Endpoint

from requestforge.interfaces import TokenProviderInterface
from requestforge.config import TokenData
from datetime import datetime, timedelta
import requests

class CustomOAuth2Provider(TokenProviderInterface):
    """Custom OAuth2 provider for proprietary API."""

    def __init__(self, token_url, client_id, client_secret, service_name):
        self._token_url = token_url
        self._client_id = client_id
        self._client_secret = client_secret
        self._service_name = service_name

    @property
    def service_name(self):
        return self._service_name

    def fetch_token(self):
        # Custom token fetch logic
        response = requests.post(
            self._token_url,
            json={
                'client_id': self._client_id,
                'client_secret': self._client_secret,
                'grant_type': 'custom_grant'
            },
            headers={'Content-Type': 'application/json'}
        )

        if response.status_code != 200:
            raise AuthenticationException(
                f'Token fetch failed: {response.status_code}',
                service_name=self._service_name
            )

        data = response.json()

        return TokenData(
            access_token=data['access_token'],
            token_type=data.get('token_type', 'Bearer'),
            expires_at=datetime.now() + timedelta(seconds=data['expires_in']),
            refresh_token=data.get('refresh_token'),
            scope=data.get('scope')
        )

    def refresh_token(self, current_token):
        if current_token.refresh_token:
            # Use refresh token
            response = requests.post(
                self._token_url,
                json={
                    'grant_type': 'refresh_token',
                    'refresh_token': current_token.refresh_token,
                    'client_id': self._client_id,
                    'client_secret': self._client_secret
                }
            )

            if response.status_code == 200:
                data = response.json()
                return TokenData.from_response(data)

        # Fallback to fetching new token
        return self.fetch_token()

# Usage
provider = CustomOAuth2Provider(
    token_url='https://auth.example.com/token',
    client_id='client-id',
    client_secret='client-secret',
    service_name='custom-api'
)

token_manager = TokenManager(provider)

Multiple Services

Different Tokens for Different Services

from requestforge import (
    HttpClient,
    HttpClientConfigBuilder,
    TokenManager,
    ClientCredentialsTokenProvider
)

# Service A
service_a_provider = ClientCredentialsTokenProvider(
    token_url='https://auth-a.example.com/token',
    client_id='service-a-id',
    client_secret='service-a-secret',
    service_name='service-a'
)
service_a_manager = TokenManager(service_a_provider)

# Service B
service_b_provider = ClientCredentialsTokenProvider(
    token_url='https://auth-b.example.com/token',
    client_id='service-b-id',
    client_secret='service-b-secret',
    service_name='service-b'
)
service_b_manager = TokenManager(service_b_provider)

# Client for Service A
client_a = HttpClient(
    HttpClientConfigBuilder()
    .with_base_url('https://api-a.example.com')
    .with_token_auth(token_manager=service_a_manager)
    .build()
)

# Client for Service B
client_b = HttpClient(
    HttpClientConfigBuilder()
    .with_base_url('https://api-b.example.com')
    .with_token_auth(token_manager=service_b_manager)
    .build()
)

# Each client uses its own token
response_a = client_a.get('/data')
response_b = client_b.get('/data')

Path Exclusion

Exclude Paths from Authentication

from requestforge import HttpClientConfigBuilder

config = (
    HttpClientConfigBuilder()
    .with_base_url('https://api.example.com')
    .with_token_auth(
        token_manager=token_manager,
        excluded_paths={
            '/health',      # Health check
            '/status',      # Status endpoint
            '/public/docs'  # Public documentation
        }
    )
    .build()
)

client = HttpClient(config)

# No auth header
response = client.get('/health')

# With auth header
response = client.get('/protected')

Pattern-Based Exclusion

config = (
    HttpClientConfigBuilder()
    .with_base_url('https://api.example.com')
    .with_token_auth(
        token_manager=token_manager,
        excluded_paths={'/health'},
        excluded_path_patterns=[
            '/public/*',        # All public paths
            '/docs/*',          # All documentation
            '*/health',         # Health at any level
            '/api/v*/status'    # Status for any version
        ]
    )
    .build()
)

Token Inspection

Accessing Token Information

from requestforge import TokenManager

token_manager = TokenManager(provider)

# Get current token
token = token_manager.get_token()

print(f"Access Token: {token.access_token}")
print(f"Token Type: {token.token_type}")
print(f"Expires At: {token.expires_at}")
print(f"Refresh Token: {token.refresh_token}")
print(f"Scope: {token.scope}")
print(f"Is Expired: {token.is_expired}")

# Get authorization header value
auth_header = token.authorization_header
print(f"Authorization: {auth_header}")

Complete OAuth2 Example

Full OAuth2 Integration

from requestforge import (
    HttpClient,
    HttpClientConfigBuilder,
    TokenManager,
    ClientCredentialsTokenProvider,
    DjangoCacheTokenStorage,
    SimpleAuthRetryStrategy,
    ExponentialBackoffRetryStrategy
)
import os

# OAuth2 provider
provider = ClientCredentialsTokenProvider(
    token_url=os.getenv('OAUTH_TOKEN_URL'),
    client_id=os.getenv('OAUTH_CLIENT_ID'),
    client_secret=os.getenv('OAUTH_CLIENT_SECRET'),
    service_name='my-api',
    scope='read write delete'
)

# Token manager with Django cache
token_manager = TokenManager(
    provider=provider,
    storage=DjangoCacheTokenStorage(
        cache_alias='default',
        key_prefix='api_oauth_tokens'
    )
)

# Auth retry strategy
auth_retry = SimpleAuthRetryStrategy(
    max_retries=1,
    delay=0.5
)

# Request retry strategy
request_retry = ExponentialBackoffRetryStrategy(
    max_retries=3,
    base_delay=1.0,
    max_delay=60.0
)

# Configure client
config = (
    HttpClientConfigBuilder()
    .with_base_url(os.getenv('API_BASE_URL'))
    .with_timeout(30.0)
    .with_retry_strategy(request_retry)
    .with_token_auth(
        token_manager=token_manager,
        auth_retry_strategy=auth_retry,
        excluded_paths={'/health', '/metrics'},
        excluded_path_patterns=['/public/*']
    )
    .with_logging(
        log_headers=True,
        log_body=False,
        sensitive_keys={'authorization'}
    )
    .with_pool_connection(20)
    .with_pool_maxsize(50)
    .build()
)

client = HttpClient(config)

# Usage
try:
    # Get users (token automatically managed)
    response = client.get('/users')
    users = response.json()

    # Create user
    response = client.post('/users', json_data={
        'name': 'John Doe',
        'email': 'john@example.com'
    })

    # Update user
    response = client.put('/users/123', json_data={
        'name': 'Jane Doe'
    })

    # Delete user
    response = client.delete('/users/123')

except HttpClientException as e:
    print(f"API error: {e}")

Error Handling

Handling OAuth2 Errors

from requestforge import (
    AuthenticationException,
    UnauthorizedException,
    HttpClientException
)

try:
    response = client.get('/protected-resource')

except AuthenticationException as e:
    print(f"Authentication failed: {e.message}")
    print(f"Service: {e.service_name}")
    # Clear cached credentials or redirect to login

except UnauthorizedException:
    print("Unauthorized - token may have been revoked")

except HttpClientException as e:
    print(f"Request failed: {e}")

Token Revocation

from requestforge import TokenManager

token_manager = TokenManager(provider)

# Invalidate token (force refresh on next use)
token_manager.invalidate_token()

# Next call will fetch new token
token = token_manager.get_token()

Testing with OAuth2

Mocking OAuth2 Tokens

import pytest
from unittest.mock import Mock
from requestforge import TokenManager, TokenData
from datetime import datetime, timedelta

def test_oauth2_request():
    # Mock provider
    mock_provider = Mock()
    mock_provider.fetch_token.return_value = TokenData(
        access_token='test-token',
        token_type='Bearer',
        expires_at=datetime.now() + timedelta(hours=1)
    )
    mock_provider.service_name = 'test-service'

    # Create token manager with mock
    token_manager = TokenManager(mock_provider)

    # Configure client
    config = (
        HttpClientConfigBuilder()
        .with_base_url('https://api.example.com')
        .with_token_auth(token_manager=token_manager)
        .build()
    )

    client = HttpClient(config)

    # Token provider called
    token = token_manager.get_token()
    assert token.access_token == 'test-token'
    mock_provider.fetch_token.assert_called_once()

OAuth2 Flow Examples

Client Credentials Flow

Basic Client Credentials

With Django Cache

With Header Authentication

With Extra Parameters

Password Grant Flow

User Authentication

Token Refresh

Automatic Refresh

Manual Refresh

Auth Retry on 401

Automatic Token Refresh on 401

Exponential Auth Retry

Custom OAuth2 Provider

Custom Token Endpoint

Multiple Services

Different Tokens for Different Services

Path Exclusion

Exclude Paths from Authentication

Pattern-Based Exclusion

Token Inspection

Accessing Token Information

Complete OAuth2 Example

Full OAuth2 Integration

Error Handling

Handling OAuth2 Errors

Token Revocation

Testing with OAuth2

Mocking OAuth2 Tokens

See Also