Parameters

def get_parameter(
    name: str,
    decrypt: bool = True,
    max_age: int = 5,
    transform: str = None,
    force_fetch: bool = False,
    **sdk_options,
) -> str:
    """
    Retrieve a single parameter from Systems Manager Parameter Store.
    
    Parameters:
    - name: Parameter name or path
    - decrypt: Whether to decrypt SecureString parameters
    - max_age: Cache TTL in seconds (0 to disable caching)
    - transform: Transformation to apply (json, base64, auto)
    - force_fetch: Whether to bypass cache and fetch fresh value
    - **sdk_options: Additional boto3 get_parameter arguments
    
    Returns:
    Parameter value as string (or transformed type if transform specified)
    
    Raises:
    GetParameterError: If parameter retrieval fails
    TransformParameterError: If transformation fails
    """

def set_parameter(
    name: str,
    value: str,
    parameter_type: str = "String",
    overwrite: bool = True,
    **sdk_options,
) -> Dict[str, Any]:
    """
    Set a parameter in Systems Manager Parameter Store.
    
    Parameters:
    - name: Parameter name
    - value: Parameter value
    - parameter_type: Parameter type (String, StringList, SecureString)
    - overwrite: Whether to overwrite existing parameter
    - **sdk_options: Additional boto3 put_parameter arguments
    
    Returns:
    Parameter version and tier information
    """

def get_parameters(
    path: str,
    recursive: bool = True,
    decrypt: bool = True,
    max_age: int = 5,
    transform: str = None,
    force_fetch: bool = False,
    **sdk_options,
) -> Dict[str, Any]:
    """
    Retrieve multiple parameters by path from Parameter Store.
    
    Parameters:
    - path: Parameter path prefix
    - recursive: Whether to retrieve parameters recursively
    - decrypt: Whether to decrypt SecureString parameters
    - max_age: Cache TTL in seconds
    - transform: Transformation to apply to all parameters
    - force_fetch: Whether to bypass cache
    - **sdk_options: Additional boto3 get_parameters_by_path arguments
    
    Returns:
    Dictionary mapping parameter names to values
    """

def get_parameters_by_name(
    parameters: List[str],
    decrypt: bool = True,
    max_age: int = 5,
    transform: str = None,
    force_fetch: bool = False,
    **sdk_options,
) -> Dict[str, Any]:
    """
    Retrieve multiple parameters by name from Parameter Store.
    
    Parameters:
    - parameters: List of parameter names
    - decrypt: Whether to decrypt SecureString parameters
    - max_age: Cache TTL in seconds
    - transform: Transformation to apply
    - force_fetch: Whether to bypass cache
    - **sdk_options: Additional boto3 get_parameters arguments
    
    Returns:
    Dictionary mapping parameter names to values
    """

def get_secret(
    name: str,
    version_id: str = None,
    version_stage: str = None,
    max_age: int = 5,
    transform: str = None,
    force_fetch: bool = False,
    **sdk_options,
) -> str:
    """
    Retrieve a secret from AWS Secrets Manager.
    
    Parameters:
    - name: Secret name or ARN
    - version_id: Specific version ID to retrieve
    - version_stage: Version stage to retrieve (AWSCURRENT, AWSPENDING)
    - max_age: Cache TTL in seconds
    - transform: Transformation to apply (json, base64, auto)
    - force_fetch: Whether to bypass cache
    - **sdk_options: Additional boto3 get_secret_value arguments
    
    Returns:
    Secret value as string (or transformed type)
    
    Raises:
    GetParameterError: If secret retrieval fails
    TransformParameterError: If transformation fails
    """

def get_secrets_by_name(
    secrets: List[str],
    max_age: int = 5,
    transform: str = None,
    force_fetch: bool = False,
    **sdk_options,
) -> Dict[str, Any]:
    """
    Retrieve multiple secrets by name from Secrets Manager.
    
    Parameters:
    - secrets: List of secret names or ARNs
    - max_age: Cache TTL in seconds
    - transform: Transformation to apply
    - force_fetch: Whether to bypass cache
    - **sdk_options: Additional boto3 arguments
    
    Returns:
    Dictionary mapping secret names to values
    """

def set_secret(
    name: str,
    secret: str,
    version_stage: str = None,
    **sdk_options,
) -> Dict[str, Any]:
    """
    Create or update a secret in AWS Secrets Manager.
    
    Parameters:
    - name: Secret name or ARN
    - secret: Secret value (string or JSON)
    - version_stage: Version stage to set
    - **sdk_options: Additional boto3 arguments
    
    Returns:
    Secret creation/update response
    """

def get_app_config(
    name: str,
    environment: str,
    application: str,
    max_age: int = 5,
    transform: str = None,
    force_fetch: bool = False,
    **sdk_options,
) -> Union[bytes, str]:
    """
    Retrieve configuration from AWS AppConfig.
    
    Parameters:
    - name: Configuration profile name
    - environment: AppConfig environment
    - application: AppConfig application name
    - max_age: Cache TTL in seconds
    - transform: Transformation to apply (json, base64, auto)
    - force_fetch: Whether to bypass cache
    - **sdk_options: Additional AppConfig arguments
    
    Returns:
    Configuration data as bytes or string (or transformed type)
    
    Raises:
    GetParameterError: If configuration retrieval fails
    TransformParameterError: If transformation fails
    """

def clear_caches() -> None:
    """
    Clear all parameter caches across all providers.
    
    Use this to force fresh retrieval of all cached parameters.
    """

class BaseProvider:
    """Base provider interface for parameter retrieval"""
    
    def __init__(
        self,
        config: Dict[str, Any] = None,
    ):
        """
        Initialize base provider.
        
        Parameters:
        - config: Provider-specific configuration
        """
    
    def get(
        self,
        name: str,
        max_age: int = None,
        transform: str = None,
        force_fetch: bool = False,
        **kwargs,
    ) -> Any:
        """
        Retrieve single parameter.
        
        Parameters:
        - name: Parameter identifier
        - max_age: Cache TTL in seconds
        - transform: Transformation to apply
        - force_fetch: Whether to bypass cache
        - **kwargs: Provider-specific arguments
        
        Returns:
        Parameter value
        """
    
    def get_multiple(
        self,
        path: str = None,
        names: List[str] = None,
        max_age: int = None,
        transform: str = None,
        force_fetch: bool = False,
        **kwargs,
    ) -> Dict[str, Any]:
        """
        Retrieve multiple parameters.
        
        Parameters:
        - path: Parameter path prefix (for hierarchical providers)
        - names: List of parameter names
        - max_age: Cache TTL in seconds
        - transform: Transformation to apply
        - force_fetch: Whether to bypass cache
        - **kwargs: Provider-specific arguments
        
        Returns:
        Dictionary of parameter name to value mappings
        """
    
    def set(
        self,
        name: str,
        value: Any,
        **kwargs,
    ) -> Any:
        """
        Set parameter value.
        
        Parameters:
        - name: Parameter identifier
        - value: Parameter value
        - **kwargs: Provider-specific arguments
        
        Returns:
        Set operation response
        """

class SSMProvider(BaseProvider):
    """Systems Manager Parameter Store provider"""
    
    def __init__(
        self,
        config: Dict[str, Any] = None,
    ):
        """
        Initialize SSM provider.
        
        Parameters:
        - config: SSM-specific configuration including boto3 session options
        """
    
    def get(
        self,
        name: str,
        decrypt: bool = True,
        max_age: int = 5,
        transform: str = None,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Any:
        """Get single parameter from SSM Parameter Store"""
    
    def get_multiple(
        self,
        path: str = None,
        names: List[str] = None,
        recursive: bool = True,
        decrypt: bool = True,
        max_age: int = 5,
        transform: str = None,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Dict[str, Any]:
        """Get multiple parameters from SSM Parameter Store"""
    
    def set(
        self,
        name: str,
        value: str,
        parameter_type: str = "String",
        overwrite: bool = True,
        **sdk_options,
    ) -> Dict[str, Any]:
        """Set parameter in SSM Parameter Store"""

class SecretsProvider(BaseProvider):
    """AWS Secrets Manager provider"""
    
    def __init__(
        self,
        config: Dict[str, Any] = None,
    ):
        """
        Initialize Secrets Manager provider.
        
        Parameters:
        - config: Secrets Manager-specific configuration
        """
    
    def get(
        self,
        name: str,
        version_id: str = None,
        version_stage: str = None,
        max_age: int = 5,
        transform: str = None,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Any:
        """Get secret from Secrets Manager"""
    
    def get_multiple(
        self,
        names: List[str],
        max_age: int = 5,
        transform: str = None,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Dict[str, Any]:
        """Get multiple secrets from Secrets Manager"""
    
    def set(
        self,
        name: str,
        secret: str,
        version_stage: str = None,
        **sdk_options,
    ) -> Dict[str, Any]:
        """Create or update secret in Secrets Manager"""

class AppConfigProvider(BaseProvider):
    """AWS AppConfig provider"""
    
    def __init__(
        self,
        environment: str,
        application: str,
        config: Dict[str, Any] = None,
    ):
        """
        Initialize AppConfig provider.
        
        Parameters:
        - environment: AppConfig environment name
        - application: AppConfig application name
        - config: AppConfig-specific configuration
        """
    
    def get(
        self,
        name: str,
        max_age: int = 5,
        transform: str = None,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Union[bytes, str]:
        """Get configuration from AppConfig"""

class DynamoDBProvider(BaseProvider):
    """DynamoDB provider for parameter storage"""
    
    def __init__(
        self,
        table_name: str,
        key_attr: str = "id",
        value_attr: str = "value",
        config: Dict[str, Any] = None,
    ):
        """
        Initialize DynamoDB provider.
        
        Parameters:
        - table_name: DynamoDB table name
        - key_attr: Attribute name for parameter key
        - value_attr: Attribute name for parameter value
        - config: DynamoDB-specific configuration
        """
    
    def get(
        self,
        name: str,
        max_age: int = 5,
        transform: str = None,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Any:
        """Get parameter from DynamoDB table"""
    
    def get_multiple(
        self,
        names: List[str],
        max_age: int = 5,
        transform: str = None,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Dict[str, Any]:
        """Get multiple parameters from DynamoDB table"""
    
    def set(
        self,
        name: str,
        value: Any,
        **sdk_options,
    ) -> Dict[str, Any]:
        """Set parameter in DynamoDB table"""

from aws_lambda_powertools.utilities.parameters import (
    get_parameter, 
    get_secret, 
    get_app_config
)
from aws_lambda_powertools.utilities.typing import LambdaContext
import json

def lambda_handler(event: dict, context: LambdaContext) -> dict:
    # Get database URL from Parameter Store
    db_url = get_parameter("/myapp/database/url", decrypt=True)
    
    # Get API key from Secrets Manager with JSON transformation
    api_credentials = get_secret("myapp/api-credentials", transform="json")
    api_key = api_credentials["api_key"]
    
    # Get feature flags from AppConfig
    feature_config = get_app_config(
        name="feature-flags",
        environment="production",
        application="myapp",
        transform="json"
    )
    
    # Use retrieved configuration
    if feature_config.get("new_feature_enabled", False):
        result = new_feature_logic(db_url, api_key)
    else:
        result = legacy_logic(db_url, api_key)
    
    return {
        "statusCode": 200,
        "body": json.dumps(result)
    }

def new_feature_logic(db_url: str, api_key: str) -> dict:
    # New feature implementation
    return {"message": "New feature executed", "version": "2.0"}

def legacy_logic(db_url: str, api_key: str) -> dict:
    # Legacy implementation
    return {"message": "Legacy feature executed", "version": "1.0"}

from aws_lambda_powertools.utilities.parameters import (
    get_parameters, 
    get_parameters_by_name,
    get_secrets_by_name
)
from aws_lambda_powertools.utilities.typing import LambdaContext

def lambda_handler(event: dict, context: LambdaContext) -> dict:
    # Get all parameters under a path
    app_config = get_parameters(
        path="/myapp/config/",
        recursive=True,
        decrypt=True,
        transform="auto"  # Auto-detect JSON/base64
    )
    
    # Get specific parameters by name
    database_params = get_parameters_by_name(
        parameters=[
            "/myapp/database/host",
            "/myapp/database/port", 
            "/myapp/database/name"
        ],
        decrypt=True
    )
    
    # Get multiple secrets
    secrets = get_secrets_by_name(
        secrets=[
            "myapp/database-credentials",
            "myapp/external-api-keys"
        ],
        transform="json"
    )
    
    # Build database connection string
    db_host = database_params["/myapp/database/host"]
    db_port = database_params["/myapp/database/port"]
    db_name = database_params["/myapp/database/name"]
    db_creds = secrets["myapp/database-credentials"]
    
    connection_string = f"postgresql://{db_creds['username']}:{db_creds['password']}@{db_host}:{db_port}/{db_name}"
    
    return {
        "statusCode": 200,
        "body": "Configuration loaded successfully"
    }

from aws_lambda_powertools.utilities.parameters import (
    SSMProvider, 
    SecretsProvider, 
    AppConfigProvider,
    clear_caches
)
from aws_lambda_powertools.utilities.typing import LambdaContext
import boto3

# Initialize custom providers with specific configurations
ssm_provider = SSMProvider(config={
    "boto3_session": boto3.Session(region_name="us-west-2")
})

secrets_provider = SecretsProvider(config={
    "boto3_session": boto3.Session(region_name="us-west-2")
})

appconfig_provider = AppConfigProvider(
    environment="production",
    application="myapp"
)

def lambda_handler(event: dict, context: LambdaContext) -> dict:
    # Use providers directly for more control
    
    # Get encrypted parameter with custom caching
    sensitive_config = ssm_provider.get(
        name="/myapp/sensitive-config",
        decrypt=True,
        max_age=300,  # Cache for 5 minutes
        transform="json"
    )
    
    # Get secret with specific version
    api_key = secrets_provider.get(
        name="myapp/api-key",
        version_stage="AWSCURRENT",
        max_age=3600  # Cache for 1 hour
    )
    
    # Get AppConfig with custom polling
    feature_flags = appconfig_provider.get(
        name="feature-flags",
        max_age=60,  # Poll every minute
        transform="json"
    )
    
    # Process based on feature flags
    results = []
    
    if feature_flags.get("enable_advanced_processing", False):
        results.append(advanced_processing(sensitive_config, api_key))
    
    if feature_flags.get("enable_reporting", False):
        results.append(generate_report(sensitive_config))
    
    # Clear caches if needed (e.g., for testing)
    if event.get("clear_cache", False):
        clear_caches()
    
    return {
        "statusCode": 200,
        "results": results
    }

def advanced_processing(config: dict, api_key: str) -> dict:
    """Advanced processing using configuration"""
    return {
        "type": "advanced",
        "processed_items": config.get("batch_size", 100),
        "api_version": "v2"
    }

def generate_report(config: dict) -> dict:
    """Generate report based on configuration"""
    return {
        "type": "report",
        "format": config.get("report_format", "json"),
        "timestamp": "2024-01-01T00:00:00Z"
    }

from aws_lambda_powertools.utilities.parameters import (
    get_parameters,
    get_secret,
    get_app_config
)
from aws_lambda_powertools.utilities.typing import LambdaContext
from typing import Dict, Any
import os

class ConfigManager:
    """Centralized configuration management"""
    
    def __init__(self):
        self._config_cache = {}
        self.app_name = os.environ.get("APP_NAME", "myapp")
        self.environment = os.environ.get("ENVIRONMENT", "dev")
    
    def get_database_config(self) -> Dict[str, Any]:
        """Get database configuration"""
        if "database" not in self._config_cache:
            # Get database parameters
            db_params = get_parameters(
                path=f"/{self.app_name}/{self.environment}/database/",
                recursive=True,
                decrypt=True
            )
            
            # Get database credentials
            db_secret = get_secret(
                name=f"{self.app_name}/{self.environment}/database-credentials",
                transform="json"
            )
            
            self._config_cache["database"] = {
                **db_params,
                "credentials": db_secret
            }
        
        return self._config_cache["database"]
    
    def get_api_config(self) -> Dict[str, Any]:
        """Get external API configuration"""
        if "api" not in self._config_cache:
            # Get API settings from Parameter Store
            api_params = get_parameters(
                path=f"/{self.app_name}/{self.environment}/api/",
                recursive=True
            )
            
            # Get API keys from Secrets Manager
            api_secrets = get_secret(
                name=f"{self.app_name}/{self.environment}/api-keys",
                transform="json"
            )
            
            self._config_cache["api"] = {
                **api_params,
                "keys": api_secrets
            }
        
        return self._config_cache["api"]
    
    def get_feature_flags(self) -> Dict[str, Any]:
        """Get feature flags from AppConfig"""
        if "features" not in self._config_cache:
            features = get_app_config(
                name="feature-flags",
                environment=self.environment,
                application=self.app_name,
                transform="json",
                max_age=30  # Refresh every 30 seconds
            )
            
            self._config_cache["features"] = features
        
        return self._config_cache["features"]
    
    def refresh_cache(self) -> None:
        """Force refresh of all cached configuration"""
        self._config_cache.clear()

# Global configuration manager instance
config_manager = ConfigManager()

def lambda_handler(event: dict, context: LambdaContext) -> dict:
    # Get configuration through manager
    db_config = config_manager.get_database_config()
    api_config = config_manager.get_api_config()
    features = config_manager.get_feature_flags()
    
    # Use configuration
    database_url = build_database_url(db_config)
    
    results = []
    
    if features.get("enable_data_processing", False):
        results.append(process_data(database_url))
    
    if features.get("enable_external_api", False):
        external_data = call_external_api(api_config)
        results.append(external_data)
    
    # Refresh config cache if requested
    if event.get("refresh_config", False):
        config_manager.refresh_cache()
    
    return {
        "statusCode": 200,
        "processed": len(results),
        "results": results
    }

def build_database_url(config: Dict[str, Any]) -> str:
    """Build database connection URL from configuration"""
    host = config.get(f"/{config_manager.app_name}/{config_manager.environment}/database/host")
    port = config.get(f"/{config_manager.app_name}/{config_manager.environment}/database/port")
    database = config.get(f"/{config_manager.app_name}/{config_manager.environment}/database/name")
    
    credentials = config["credentials"]
    username = credentials["username"]
    password = credentials["password"]
    
    return f"postgresql://{username}:{password}@{host}:{port}/{database}"

def call_external_api(config: Dict[str, Any]) -> Dict[str, Any]:
    """Call external API using configuration"""
    import requests
    
    base_url = config.get(f"/{config_manager.app_name}/{config_manager.environment}/api/base_url")
    timeout = int(config.get(f"/{config_manager.app_name}/{config_manager.environment}/api/timeout", "30"))
    
    api_key = config["keys"]["primary_api_key"]
    
    response = requests.get(
        f"{base_url}/data",
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=timeout
    )
    
    return response.json()

from aws_lambda_powertools.utilities.parameters import (
    get_parameter,
    get_secret, 
    GetParameterError,
    TransformParameterError
)
from aws_lambda_powertools.utilities.typing import LambdaContext
import time
import json

def lambda_handler(event: dict, context: LambdaContext) -> dict:
    try:
        # Get configuration with retry logic
        config = get_configuration_with_retry()
        
        # Process event using configuration
        result = process_event(event, config)
        
        return {
            "statusCode": 200,
            "body": json.dumps(result)
        }
        
    except GetParameterError as e:
        print(f"Failed to retrieve parameter: {str(e)}")
        return {
            "statusCode": 500,
            "body": json.dumps({"error": "Configuration unavailable"})
        }
    
    except TransformParameterError as e:
        print(f"Failed to transform parameter: {str(e)}")
        return {
            "statusCode": 500,
            "body": json.dumps({"error": "Configuration format invalid"})
        }

def get_configuration_with_retry(max_retries: int = 3, retry_delay: float = 1.0) -> dict:
    """Get configuration with exponential backoff retry"""
    
    for attempt in range(max_retries):
        try:
            # Attempt to get all required configuration
            config = {}
            
            # Get database configuration
            config["database_url"] = get_parameter(
                "/myapp/database/url",
                decrypt=True,
                force_fetch=True  # Always get fresh value on retry
            )
            
            # Get API configuration
            api_config = get_secret(
                "myapp/api-config",
                transform="json",
                force_fetch=True
            )
            config["api"] = api_config
            
            # Get feature flags
            feature_flags = get_parameter(
                "/myapp/feature-flags",
                transform="json",
                force_fetch=True
            )
            config["features"] = feature_flags
            
            return config
            
        except (GetParameterError, TransformParameterError) as e:
            if attempt == max_retries - 1:
                # Last attempt failed, re-raise
                raise
            
            # Wait before retrying with exponential backoff
            wait_time = retry_delay * (2 ** attempt)
            print(f"Configuration retrieval failed (attempt {attempt + 1}/{max_retries}): {str(e)}")
            print(f"Retrying in {wait_time} seconds...")
            time.sleep(wait_time)

def process_event(event: dict, config: dict) -> dict:
    """Process event using retrieved configuration"""
    
    # Use database connection
    database_url = config["database_url"]
    
    # Use API configuration
    api_endpoint = config["api"]["endpoint"]
    api_key = config["api"]["key"]
    
    # Check feature flags
    features = config["features"]
    
    result = {
        "processed_at": time.time(),
        "event_id": event.get("id", "unknown"),
        "features_enabled": []
    }
    
    if features.get("enable_processing", False):
        result["features_enabled"].append("processing")
        # Process event data
        result["processed_records"] = len(event.get("records", []))
    
    if features.get("enable_notifications", False):
        result["features_enabled"].append("notifications")
        # Send notifications
        result["notifications_sent"] = 1
    
    return result

from typing import Dict, Any, List, Union, Optional, Callable
import boto3

# Parameter transformation types
TransformType = Literal["json", "base64", "auto"]

# Parameter provider configuration
ProviderConfig = Dict[str, Any]

# Parameter retrieval options
ParameterOptions = Dict[str, Any]

# Exception types
class GetParameterError(Exception):
    """Raised when parameter retrieval fails"""
    def __init__(self, message: str, parameter_name: str = None): ...

class TransformParameterError(Exception):
    """Raised when parameter transformation fails"""
    def __init__(self, message: str, parameter_name: str = None, transform: str = None): ...

# Provider response types
ParameterValue = Union[str, int, float, bool, Dict[str, Any], List[Any], bytes]
ParameterDict = Dict[str, ParameterValue]

# Boto3 session type for provider configuration
Boto3Session = boto3.Session

# Cache configuration
class CacheConfig:
    def __init__(
        self,
        max_age: int = 5,
        max_size: int = 1000,
        enabled: bool = True,
    ):
        """
        Cache configuration for parameter providers.
        
        Parameters:
        - max_age: Default cache TTL in seconds
        - max_size: Maximum number of cached items
        - enabled: Whether caching is enabled
        """

# SDK options for AWS service calls
SSMOptions = Dict[str, Any]
SecretsManagerOptions = Dict[str, Any] 
AppConfigOptions = Dict[str, Any]
DynamoDBOptions = Dict[str, Any]