#!/usr/bin/env python3
"""
###########################################################################################################
# File: amorphic_api_client.py
# Location: /amorphic_client/amorphic_api_client.py
#
# This module provides the AmorphicApiClient class which extends the base OpenAPI ApiClient
# to provide enhanced functionality for interacting with the Amorphic Data Platform APIs.
# It includes system information retrieval, API version compatibility checking, authentication,
# role management, and comprehensive logging capabilities with optional custom logger support.
#
# Modification History:
# ===================================================================
# Date Who Description
# ========== ================= ==============================
#
# Jun 2025 Subir Adhikari Initial version of Amorphic SDK client
# with logging, version checking, and
# optional custom logger support
#
#
###########################################################################################################
"""
import json
import logging
import os
from typing import Optional, Any
import certifi
from packaging.version import parse as version_parse
import boto3
from botocore.exceptions import ClientError, NoCredentialsError
from openapi_client.api_client import ApiClient
from openapi_client.configuration import Configuration
from openapi_client.api.management_api import ManagementApi
from openapi_client.rest import ApiException
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
[docs]
class TokenFetcher:
"""Utility class to fetch PAT tokens from various sources for local development"""
[docs]
def __init__(self, aws_profile: Optional[str] = None, logger: Optional[logging.Logger] = None):
"""
Initialize TokenFetcher
Args:
aws_profile (Optional[str]): AWS profile name for local development
logger (Optional[logging.Logger]): Logger instance for consistent logging
"""
self.aws_profile = aws_profile
self.logger = logger or logging.getLogger(__name__)
def _get_boto3_client(self, service_name: str) -> Any:
"""
Get boto3 client with appropriate configuration
Args:
service_name (str): AWS service name (e.g., 'ssm', 'secretsmanager')
Returns:
boto3 client instance
Raises:
NoCredentialsError: If AWS credentials are not configured
"""
try:
if self.aws_profile:
self.logger.info("Creating %s client with profile: %s", service_name, self.aws_profile)
session = boto3.Session(profile_name=self.aws_profile)
return session.client(service_name)
self.logger.info("Creating %s client with default credentials", service_name)
return boto3.client(service_name)
except NoCredentialsError as e:
self.logger.error("AWS credentials not configured. Please run 'aws configure' or set environment variables")
raise NoCredentialsError(
"AWS credentials not configured. Please run 'aws configure' or set environment variables"
) from e
[docs]
def fetch_token_from_env(self, env_var_name: str) -> Optional[str]:
"""
Fetch token from environment variable
Args:
env_var_name (str): Environment variable name
Returns:
Optional[str]: Token if found, None otherwise
"""
token = os.getenv(env_var_name)
if token:
self.logger.info("Token retrieved from environment variable: %s", env_var_name)
return token
self.logger.warning("Environment variable %s not found", env_var_name)
return None
[docs]
def fetch_token_from_ssm(self, parameter_name: str, decrypt: bool = True) -> Optional[str]:
"""
Fetch token from AWS SSM Parameter Store
Args:
parameter_name (str): SSM parameter name
decrypt (bool): Whether to decrypt SecureString parameters
Returns:
Optional[str]: Token if found, None otherwise
"""
try:
self.logger.debug("Attempting to fetch token from SSM parameter: %s", parameter_name)
ssm_client = self._get_boto3_client('ssm')
response = ssm_client.get_parameter(
Name=parameter_name,
WithDecryption=decrypt
)
token = response['Parameter']['Value']
self.logger.info("Successfully retrieved token from SSM parameter: %s", parameter_name)
return token
except ClientError as e:
error_code = e.response['Error']['Code']
if error_code == 'ParameterNotFound':
self.logger.warning("SSM parameter not found: %s", parameter_name)
elif error_code == 'AccessDenied':
self.logger.error("Access denied to SSM parameter: %s", parameter_name)
else:
self.logger.error("Error retrieving SSM parameter %s: %s", parameter_name, e)
return None
except Exception as e:
self.logger.error("Unexpected error retrieving SSM parameter %s: %s", parameter_name, e)
return None
[docs]
def fetch_token_from_secrets_manager(self, secret_arn: str) -> Optional[str]:
"""
Fetch token from AWS Secrets Manager
Args:
secret_arn (str): Secret ARN or name
Returns:
Optional[str]: Token if found, None otherwise
"""
try:
self.logger.debug("Attempting to fetch token from Secrets Manager: %s", secret_arn)
secrets_client = self._get_boto3_client('secretsmanager')
response = secrets_client.get_secret_value(SecretId=secret_arn)
if 'SecretString' in response:
secret_value = response['SecretString']
self.logger.info("Successfully retrieved token from Secrets Manager: %s", secret_arn)
return secret_value
self.logger.error("Secret does not contain string value: %s", secret_arn)
return None
except ClientError as e:
error_code = e.response['Error']['Code']
if error_code == 'ResourceNotFoundException':
self.logger.warning("Secret not found: %s", secret_arn)
elif error_code == 'AccessDenied':
self.logger.error("Access denied to secret: %s", secret_arn)
else:
self.logger.error("Error retrieving secret %s: %s", secret_arn, e)
return None
except Exception as e:
self.logger.error("Unexpected error retrieving secret %s: %s", secret_arn, e)
return None
[docs]
def fetch_token(self,
env_var: Optional[str] = None,
ssm_parameter: Optional[str] = None,
secret_arn: Optional[str] = None) -> str:
"""
Fetch token from the specified source
Args:
env_var (Optional[str]): Environment variable name
ssm_parameter (Optional[str]): SSM parameter name
secret_arn (Optional[str]): Secrets Manager secret ARN
Returns:
str: Token from the specified source
Raises:
ValueError: If no token source is specified
"""
self.logger.debug("Attempting to fetch token from configured sources")
if env_var:
self.logger.debug("Fetching token from environment variable: %s", env_var)
return self.fetch_token_from_env(env_var)
if ssm_parameter:
self.logger.debug("Fetching token from SSM parameter: %s", ssm_parameter)
return self.fetch_token_from_ssm(ssm_parameter)
if secret_arn:
self.logger.debug("Fetching token from Secrets Manager: %s", secret_arn)
return self.fetch_token_from_secrets_manager(secret_arn)
self.logger.error("No token source specified")
raise ValueError("No token source specified")
[docs]
class AmorphicApiClient(ApiClient):
"""Custom API client for Amorphic API that extends the base ApiClient"""
[docs]
def __init__(self, configuration: Configuration, role_id: str, custom_logger: Optional[logging.Logger] = None):
"""
Initialize the Amorphic API client with the given configuration and role ID
Args:
configuration (Configuration): The configuration object containing host, API key, etc.
role_id (str): The role ID for authentication
custom_logger (Optional[logging.Logger]): Custom logger instance. If not provided, a default logger will be created.
Raises:
ValueError: If the API version is not compatible with the required version range
"""
# Initialize logger - use provided logger or create default one
if custom_logger is not None:
self.logger = custom_logger
# Set log level on provided logger if debug is enabled
if configuration.debug and self.logger.level > logging.DEBUG:
self.logger.setLevel(logging.DEBUG)
else:
# Create default logger if none provided
self.logger = logging.getLogger(__name__)
if not self.logger.handlers:
handler = logging.StreamHandler()
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
# Set log level based on debug configuration
if configuration.debug:
self.logger.setLevel(logging.DEBUG)
else:
self.logger.setLevel(logging.INFO)
# Log configuration values only when debug is True
if configuration.debug:
self._log_configuration(configuration)
super().__init__(configuration)
self.management_api = ManagementApi(self)
self.role_id = role_id
try:
self._check_version_compatibility()
except ValueError:
raise # Re-raise version compatibility errors
except Exception as e:
self.logger.warning("Could not get system information during initialization: %s", e)
def _log_configuration(self, configuration: Configuration) -> None:
"""Log configuration values when debug is enabled"""
config_dict = {}
for k, v in configuration.__dict__.items():
if v is not None and k not in ['logger', 'logger_file_handler', 'debug']:
try:
# Test if value can be serialized
if isinstance(v, dict):
v_copy = v.copy()
if 'api_key' in v_copy and isinstance(v_copy['api_key'], dict):
v_copy['api_key']['LambdaAuthorizer'] = '********'
json.dumps(v_copy) # Test serialization
config_dict[k] = v_copy
else:
json.dumps({k: v}) # Test serialization
config_dict[k] = v
except (TypeError, ValueError):
continue
self.logger.debug("Configuration passed to the client: %s", json.dumps(config_dict, indent=2))
def _check_version_compatibility(self) -> None:
"""Check version compatibility with the API"""
# Load version configuration
config_path = os.path.join(os.path.dirname(__file__), 'version_config.json')
with open(config_path, 'r', encoding='utf-8') as f:
version_config = json.load(f)
# Find current SDK version configuration
current_version_config = None
for config in version_config.values():
if config.get('is_current', False):
current_version_config = config
break
if not current_version_config:
raise ValueError("No current SDK version found in version configuration")
min_version = current_version_config.get('min_amorphic_version')
max_version = current_version_config.get('max_amorphic_version')
# Get system information
system_info = self.management_api.get_system_information(role_id=self.role_id)
current_version = system_info.version
# Check version compatibility
if min_version and version_parse(current_version) < version_parse(min_version):
raise ValueError(
f"API version {current_version} is below minimum required version {min_version}"
)
if max_version and version_parse(current_version) > version_parse(max_version):
raise ValueError(
f"API version {current_version} is above maximum allowed version {max_version}"
)
self._log_system_info(system_info)
def _log_system_info(self, system_info: Any) -> None:
"""Log system information"""
self.logger.info("\nSystem Information:")
self.logger.info("=" * 50)
self.logger.info("Version: %s", system_info.version)
self.logger.info("Environment: %s", system_info.environment)
self.logger.info("Project Name: %s", system_info.project_name)
self.logger.info("Project Shortname: %s", system_info.project_shortname)
self.logger.info("AWS Region: %s", system_info.aws_region)
self.logger.info("AWS Account ID: %s", system_info.aws_account_id)
self.logger.info("-" * 50)
[docs]
@classmethod
def create_with_auth(cls,
host: str,
role_id: str,
ssl_ca_cert: Optional[str] = None,
debug: bool = False,
aws_profile: Optional[str] = None,
env_var: Optional[str] = None,
ssm_parameter: Optional[str] = None,
secret_arn: Optional[str] = None,
token: Optional[str] = None,
custom_logger: Optional[logging.Logger] = None) -> 'AmorphicApiClient':
"""
Factory method to create AmorphicApiClient with flexible authentication
Args:
host (str): API host URL
role_id (str): Role ID for authentication
ssl_ca_cert (Optional[str]): SSL CA certificate path
debug (bool): Enable debug mode
aws_profile (Optional[str]): AWS profile for local development
env_var (Optional[str]): Environment variable name containing token
ssm_parameter (Optional[str]): SSM parameter name containing token
secret_arn (Optional[str]): Secrets Manager secret ARN
token (Optional[str]): Direct token value
custom_logger (Optional[logging.Logger]): Custom logger instance
Returns:
AmorphicApiClient: Configured client instance
Raises:
ValueError: If no token source is configured
"""
if not token:
token_fetcher = TokenFetcher(aws_profile=aws_profile, logger=custom_logger)
token = token_fetcher.fetch_token(
env_var=env_var,
ssm_parameter=ssm_parameter,
secret_arn=secret_arn
)
configuration = Configuration(
host=host,
api_key={'LambdaAuthorizer': token},
ssl_ca_cert=ssl_ca_cert,
debug=debug
)
return cls(configuration=configuration, role_id=role_id, custom_logger=custom_logger)