AutoRest swagger generator Python client runtime for REST API clients with serialization, authentication, and request handling.
—
Comprehensive authentication system supporting multiple authentication schemes including Basic, Bearer token, OAuth2, API keys, Kerberos, and Azure service-specific credentials. The authentication system is designed to be pluggable and extensible for various REST API requirements.
Base authentication class providing the foundation for all authentication mechanisms.
class Authentication:
header: str = "Authorization"
def signed_session(self, session=None):
"""
Create requests session with auth headers applied.
Parameters:
- session: Optional existing requests.Session to configure
Returns:
Configured requests.Session object
"""HTTP Basic authentication using username and password credentials.
class BasicAuthentication(Authentication):
def __init__(self, username: str, password: str):
"""
Initialize Basic authentication.
Parameters:
- username: Authentication username
- password: Authentication password
"""
scheme: str = 'Basic'
username: str
password: strSimple Bearer token authentication for APIs that use static tokens.
class BasicTokenAuthentication(Authentication):
def __init__(self, token: dict):
"""
Initialize Bearer token authentication.
Parameters:
- token: Token dict with 'access_token' key
"""
scheme: str = 'Bearer'
token: dict
def set_token(self):
"""Define token attribute (no-op in basic implementation)."""OAuth2 token authentication with automatic token refresh capabilities.
class OAuthTokenAuthentication(BasicTokenAuthentication):
def __init__(self, client_id: str, token: dict):
"""
Initialize OAuth2 authentication.
Parameters:
- client_id: OAuth2 client ID
- token: OAuth2 token dict with expires_in field required
"""
id: str # OAuth2 client ID
store_key: str # Storage key for token
def construct_auth(self) -> str:
"""
Format token header string.
Note: Uses requests_oauthlib.OAuth2 instead of manual header construction
"""
def refresh_session(self, session=None):
"""
Return updated session with OAuth2 authentication configured.
Parameters:
- session: Optional existing session to update
Returns:
Updated requests.Session with OAuth2 auth
Note: Does not call parent signed_session by design
"""Kerberos Single Sign-On (SSO) authentication requiring requests_kerberos package.
class KerberosAuthentication(Authentication):
def __init__(self, mutual_authentication=None):
"""
Initialize Kerberos authentication.
Parameters:
- mutual_authentication: Mutual auth requirement (REQUIRED, OPTIONAL, DISABLED)
Raises:
- ImportError: If requests_kerberos package is not installed
"""
mutual_authentication: any # Kerberos mutual authentication settingAPI key authentication supporting keys in headers and/or query parameters.
class ApiKeyCredentials(Authentication):
def __init__(self, in_headers=None, in_query=None):
"""
Initialize API key authentication.
Parameters:
- in_headers: Dict of header-based API keys
- in_query: Dict of query parameter-based API keys
Raises:
- ValueError: If both in_headers and in_query are None or empty
- ValueError: If session.params is not a dict during signing
"""
in_headers: dict # API keys to include in request headers
in_query: dict # API keys to include in query parametersSpecialized credential classes for Azure services with predefined header formats.
class CognitiveServicesCredentials(ApiKeyCredentials):
_subscription_key_header = 'Ocp-Apim-Subscription-Key'
def __init__(self, subscription_key: str):
"""
Cognitive Services authentication.
Parameters:
- subscription_key: Azure Cognitive Services subscription key
Raises:
- ValueError: If subscription_key is None
Note: Automatically adds X-BingApis-SDK-Client header
"""
class TopicCredentials(ApiKeyCredentials):
_topic_key_header = 'aeg-sas-key'
def __init__(self, topic_key: str):
"""
Event Grid topic authentication.
Parameters:
- topic_key: Event Grid topic access key
Raises:
- ValueError: If topic_key is None
"""
class DomainCredentials(ApiKeyCredentials):
_domain_key_header = 'aeg-sas-key'
def __init__(self, domain_key: str):
"""
Event Grid domain authentication.
Parameters:
- domain_key: Event Grid domain access key
Raises:
- ValueError: If domain_key is None
"""from msrest import ServiceClient, Configuration
from msrest.authentication import BasicAuthentication
# Create basic auth
auth = BasicAuthentication('myusername', 'mypassword')
# Configure client
config = Configuration(base_url='https://api.example.com')
config.credentials = auth
# Use client
with ServiceClient(None, config) as client:
request = client.get('/protected/resource')
response = client.send(request)from msrest.authentication import ApiKeyCredentials
# API key in header
auth = ApiKeyCredentials(in_headers={'X-API-Key': 'your-api-key-here'})
# API key in query parameter
auth = ApiKeyCredentials(in_query={'api_key': 'your-api-key-here'})
# API key in both header and query
auth = ApiKeyCredentials(
in_headers={'Authorization': 'Bearer your-token'},
in_query={'client_id': 'your-client-id'}
)from msrest.authentication import OAuthTokenAuthentication
# OAuth2 token (assuming you've obtained it through OAuth flow)
token = {
'access_token': 'your-access-token',
'token_type': 'Bearer',
'expires_in': 3600,
'refresh_token': 'your-refresh-token'
}
auth = OAuthTokenAuthentication('your-client-id', token)
# The authentication will automatically handle token refresh
config = Configuration(base_url='https://api.example.com')
config.credentials = authfrom msrest.authentication import CognitiveServicesCredentials
# Use Azure Cognitive Services subscription key
auth = CognitiveServicesCredentials('your-subscription-key')
config = Configuration(base_url='https://westus.api.cognitive.microsoft.com')
config.credentials = authfrom msrest.authentication import KerberosAuthentication
# Requires: pip install requests_kerberos
try:
from requests_kerberos import REQUIRED, OPTIONAL, DISABLED
# Use Kerberos with required mutual authentication
auth = KerberosAuthentication(mutual_authentication=REQUIRED)
config = Configuration(base_url='https://api.example.com')
config.credentials = auth
except ImportError:
print("Install requests_kerberos for Kerberos support")import requests
from msrest.authentication import BasicAuthentication
# Create authentication
auth = BasicAuthentication('username', 'password')
# Get configured session
session = auth.signed_session()
# Customize session further
session.verify = False # Disable SSL verification
session.proxies = {'http': 'http://proxy.example.com:8080'}
# Use session directly or with ServiceClient
response = session.get('https://api.example.com/data')Authentication credentials can be set on the Configuration object:
from msrest import ServiceClient, Configuration
from msrest.authentication import ApiKeyCredentials
# Method 1: Set credentials on config
config = Configuration(base_url='https://api.example.com')
config.credentials = ApiKeyCredentials(in_headers={'X-API-Key': 'key'})
client = ServiceClient(None, config)
# Method 2: Pass credentials to client (deprecated)
auth = ApiKeyCredentials(in_headers={'X-API-Key': 'key'})
client = ServiceClient(auth, config)Authentication errors are handled through specific exception types:
from msrest.exceptions import AuthenticationError, TokenExpiredError
try:
with ServiceClient(None, config) as client:
response = client.send(request)
except AuthenticationError as e:
print(f"Authentication failed: {e}")
except TokenExpiredError as e:
print(f"Token expired: {e}")
# Handle token refreshInstall with Tessl CLI
npx tessl i tessl/pypi-msrest