Token Manager API Reference

This page documents the token management system including providers, storage, and managers.

Token Manager

TokenManager

class TokenManager(provider, storage=None)

Thread-safe token manager with caching.

Parameters:

provider (TokenProviderInterface) – Token provider
storage (TokenStorageInterface) – Token storage (default: InMemoryTokenStorage)

Features:

Automatic token caching
Thread-safe token refresh
Prevents concurrent token fetches
Automatic expiration handling

Thread Safety: This class is thread-safe. Multiple threads can safely share a single instance.

Example:

from requestforge import TokenManager, ClientCredentialsTokenProvider, InMemoryTokenStorage

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=provider,
    storage=InMemoryTokenStorage()
)

Methods

TokenManager.get_token()

Get valid token (from cache or fetch new).

Returns:: Valid token data
Return type:: TokenData
Raises:: AuthenticationException – If token fetch fails

Behavior:

Check cache for valid token
If cached and not expired, return cached token
If expired or not cached, fetch new token
Cache new token
Return token

Example:

token = token_manager.get_token()
print(token.access_token)
print(token.expires_at)

TokenManager.invalidate_token()

Invalidate the current cached token.

Forces next get_token() call to fetch a new token.

Example:

# Invalidate token (e.g., after logout)
token_manager.invalidate_token()

TokenManager.force_refresh()

Force fetch a new token regardless of cache.

Returns:: New token data
Return type:: TokenData
Raises:: AuthenticationException – If token fetch fails

Example:

# Force refresh token
new_token = token_manager.force_refresh()

Token Storage

TokenStorageInterface

class TokenStorageInterface

Abstract interface for token storage.

Methods:

get(key)

Retrieve token by key.

Parameters:: key (str) – Storage key
Returns:: Token data or None
Return type:: TokenData | None

set(key, token)

Store token with key.

Parameters:

key (str) – Storage key
token (TokenData) – Token to store

delete(key)

Delete token by key.

Parameters:: key (str) – Storage key

exists(key)

Check if token exists.

Parameters:: key (str) – Storage key
Returns:: True if exists
Return type:: bool

InMemoryTokenStorage

class InMemoryTokenStorage

Thread-safe in-memory token storage.

Use Cases:

Single-instance deployments
Development/testing
Applications without shared cache

Limitations:

Tokens not shared across processes
Tokens lost on restart

Example:

from requestforge import InMemoryTokenStorage

storage = InMemoryTokenStorage()
token_manager = TokenManager(provider, storage)

Methods:

get(key): Get token from memory.

set(key, token): Store token in memory.

delete(key): Delete token from memory.

exists(key): Check if token exists in memory.

clear()

Clear all tokens from memory.

Example:

storage.clear()  # Remove all cached tokens

DjangoCacheTokenStorage

class DjangoCacheTokenStorage(cache_alias='default', key_prefix='token')

Django cache-based token storage.

Parameters:

cache_alias (str) – Django cache alias
key_prefix (str) – Prefix for cache keys

Use Cases:

Multi-instance deployments
Production environments
Shared cache backends (Redis, Memcached)

Requirements:

Django installed
Configured cache backend

Example:

from requestforge import DjangoCacheTokenStorage

storage = DjangoCacheTokenStorage(
    cache_alias='default',
    key_prefix='api_tokens'
)

token_manager = TokenManager(provider, storage)

Django Settings:

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

Methods:

get(key): Get token from Django cache.

set(key, token)

Store token in Django cache with TTL.

TTL is calculated from token expiration with minimum of 60 seconds.

delete(key): Delete token from Django cache.

exists(key): Check if token exists in Django cache.

Custom Storage Example

from requestforge.interfaces import TokenStorageInterface
from requestforge.config import TokenData
import redis
import json
from datetime import datetime

class RedisTokenStorage(TokenStorageInterface):
    """Redis-based token storage."""

    def __init__(self, redis_client, key_prefix='token'):
        self.redis = redis_client
        self.prefix = key_prefix

    def _make_key(self, key):
        return f"{self.prefix}:{key}"

    def get(self, key):
        data = self.redis.get(self._make_key(key))
        if data:
            token_dict = json.loads(data)
            # Convert expires_at string to datetime
            if token_dict.get('expires_at'):
                token_dict['expires_at'] = datetime.fromisoformat(
                    token_dict['expires_at']
                )
            return TokenData(**token_dict)
        return None

    def set(self, key, token):
        token_dict = {
            'access_token': token.access_token,
            'token_type': token.token_type,
            'expires_at': token.expires_at.isoformat() if token.expires_at else None,
            'refresh_token': token.refresh_token,
            'scope': token.scope,
            'extra': token.extra
        }

        data = json.dumps(token_dict)

        # Calculate TTL
        if token.expires_at:
            ttl = int((token.expires_at - datetime.now()).total_seconds())
            ttl = max(ttl, 60)  # Minimum 60 seconds
            self.redis.setex(self._make_key(key), ttl, data)
        else:
            self.redis.set(self._make_key(key), data)

    def delete(self, key):
        self.redis.delete(self._make_key(key))

    def exists(self, key):
        return self.redis.exists(self._make_key(key))

# Usage
redis_client = redis.Redis(host='localhost', port=6379, db=0)
storage = RedisTokenStorage(redis_client)
token_manager = TokenManager(provider, storage)

Token Providers

TokenProviderInterface

class TokenProviderInterface

Abstract interface for token providers.

Properties:

service_name: str: Unique identifier for this token provider.

Methods:

fetch_token()

Fetch a new token from the authentication server.

Returns:: Fresh token data
Return type:: TokenData
Raises:: AuthenticationException – If fetch fails

refresh_token(current_token)

Refresh an existing token.

Parameters:: current_token (TokenData) – Current token
Returns:: Refreshed token data
Return type:: TokenData
Raises:: AuthenticationException – If refresh fails

BaseOAuth2TokenProvider

class BaseOAuth2TokenProvider(token_url, client_id, client_secret, service_name, scope=None, extra_params=None)

Base OAuth2 token provider.

Parameters:

token_url (str) – Token endpoint URL
client_id (str) – OAuth2 client ID
client_secret (str) – OAuth2 client secret
service_name (str) – Service identifier
scope (str) – OAuth2 scope (optional)
extra_params (dict) – Additional parameters (optional)

Note: This is an abstract base class. Use concrete implementations like ClientCredentialsTokenProvider.

Properties:

service_name: str: Service name for this provider.

Methods:

fetch_token(): Fetch new token using client credentials.

refresh_token(current_token): Refresh existing token or fetch new one.

ClientCredentialsTokenProvider

class ClientCredentialsTokenProvider(token_url, client_id, client_secret, service_name, scope=None, grant_type='client_credentials', extra_params=None, auth_method='body')

OAuth2 Client Credentials grant type token provider.

Parameters:

token_url (str) – Token endpoint URL
client_id (str) – Client ID
client_secret (str) – Client secret
service_name (str) – Service name
scope (str) – OAuth2 scope (optional)
grant_type (str) – Grant type (default: ‘client_credentials’)
extra_params (dict) – Additional parameters (optional)
auth_method (str) – Authentication method (‘body’ or ‘header’)

Auth Methods:

body: Send credentials in request body
header: Send credentials in Authorization header (HTTP Basic)

Example:

from requestforge import ClientCredentialsTokenProvider

# Body authentication (default)
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',
    auth_method='body'
)

# Header authentication
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'
)

token_manager = TokenManager(provider)

PasswordGrantTokenProvider

class PasswordGrantTokenProvider(token_url, client_id, client_secret, username, password, service_name, scope=None, extra_params=None)

OAuth2 Password grant type token provider.

Parameters:

token_url (str) – Token endpoint URL
client_id (str) – Client ID
client_secret (str) – Client secret
username (str) – User username
password (str) – User password
service_name (str) – Service name
scope (str) – OAuth2 scope (optional)
extra_params (dict) – Additional parameters (optional)

Example:

from requestforge import PasswordGrantTokenProvider

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 = TokenManager(provider)

Custom Provider Example

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

class CustomTokenProvider(TokenProviderInterface):
    """Custom token provider for proprietary API."""

    def __init__(self, api_url, api_key, service_name):
        self._api_url = api_url
        self._api_key = api_key
        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(
            f'{self._api_url}/auth/token',
            headers={'X-API-Key': self._api_key},
            json={'grant_type': 'api_key'}
        )

        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['token'],
            token_type='Custom',
            expires_at=datetime.now() + timedelta(seconds=data['ttl']),
            extra={'api_version': data.get('version')}
        )

    def refresh_token(self, current_token):
        # For this API, just fetch a new token
        return self.fetch_token()

# Usage
provider = CustomTokenProvider(
    api_url='https://api.example.com',
    api_key='your-api-key',
    service_name='custom-api'
)
token_manager = TokenManager(provider)

Complete Examples

Basic Token Management

from requestforge import (
    TokenManager,
    ClientCredentialsTokenProvider,
    InMemoryTokenStorage
)

# Create 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'
)

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

# Get token (fetched and cached)
token = token_manager.get_token()
print(f"Token: {token.access_token}")
print(f"Expires: {token.expires_at}")

# Get token again (from cache)
token = token_manager.get_token()

# Force refresh
new_token = token_manager.force_refresh()

With HTTP Client

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

# Setup token manager
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)

# Configure 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')

Django Integration

# settings.py
from requestforge import (
    TokenManager,
    ClientCredentialsTokenProvider,
    DjangoCacheTokenStorage
)

# Token provider
TOKEN_PROVIDER = ClientCredentialsTokenProvider(
    token_url='https://auth.example.com/oauth/token',
    client_id=os.getenv('OAUTH_CLIENT_ID'),
    client_secret=os.getenv('OAUTH_CLIENT_SECRET'),
    service_name='api-service'
)

# Token manager with Django cache
TOKEN_MANAGER = TokenManager(
    provider=TOKEN_PROVIDER,
    storage=DjangoCacheTokenStorage(
        cache_alias='default',
        key_prefix='api_tokens'
    )
)

# views.py
from django.conf import settings
from requestforge import HttpClient, HttpClientConfigBuilder

config = (
    HttpClientConfigBuilder()
    .with_base_url('https://api.example.com')
    .with_token_auth(token_manager=settings.TOKEN_MANAGER)
    .build()
)

client = HttpClient(config)
response = client.get('/data')

Multi-Service Tokens

from requestforge import TokenManager, ClientCredentialsTokenProvider

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

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

# Create clients with different tokens
client_a = HttpClient(
    HttpClientConfigBuilder()
    .with_base_url('https://api.service-a.com')
    .with_token_auth(token_manager=service_a_tokens)
    .build()
)

client_b = HttpClient(
    HttpClientConfigBuilder()
    .with_base_url('https://api.service-b.com')
    .with_token_auth(token_manager=service_b_tokens)
    .build()
)

Token Refresh Exception

TokenRefreshException

See requestforge.exceptions.TokenRefreshException for detailed exception documentation.

Example:

from requestforge.token_manager import TokenRefreshException

try:
    token = token_manager.get_token()
except TokenRefreshException as e:
    print(f"Token refresh failed: {e}")

Token Manager API Reference

Token Manager

TokenManager

Methods

Token Storage

TokenStorageInterface

InMemoryTokenStorage

DjangoCacheTokenStorage

Custom Storage Example

Token Providers

TokenProviderInterface

BaseOAuth2TokenProvider

ClientCredentialsTokenProvider

PasswordGrantTokenProvider

Custom Provider Example

Complete Examples

Basic Token Management

With HTTP Client

Django Integration

Multi-Service Tokens

Token Refresh Exception

TokenRefreshException

See Also