import json from typing import Any, Dict, Optional import boto3 from botocore.exceptions import ClientError def get_secret( secret_name: str, region_name: Optional[str] = "us-east-1" ) -> Dict[str, Any]: """ Retrieve a secret from AWS Secrets Manager. Args: secret_name (str): The name or ARN of the secret to retrieve region_name (Optional[str]): The AWS region name. If not provided, will try to get from environment variable AWS_REGION or default to 'us-east-1' Returns: Dict[str, Any]: The secret value as a dictionary Raises: SecretsManagerError: If there's an error retrieving the secret ValueError: If the secret_name is empty or None """ if not secret_name: raise ValueError("secret_name cannot be empty or None") try: session = boto3.session.Session() client = session.client(service_name="secretsmanager", region_name=region_name) response = client.get_secret_value(SecretId=secret_name) if "SecretString" in response: try: secret_value = json.loads(response["SecretString"]) except json.JSONDecodeError: # If not JSON, return as string secret_value = response["SecretString"] else: # Handle binary secrets secret_value = response["SecretBinary"] return secret_value except ClientError as e: error_code = e.response["Error"]["Code"] error_message = e.response["Error"]["Message"] if error_code == "ResourceNotFoundException": raise Exception(f"Secret {secret_name} not found, {e}") elif error_code == "InvalidRequestException": raise Exception(f"Invalid request for secret {secret_name}, {e}") elif error_code == "InvalidParameterException": raise Exception(f"Invalid parameter for secret {secret_name}, {e}") else: raise Exception( f"Error retrieving secret {secret_name}: {error_message}, {e}" ) except Exception as e: raise Exception(f"Unexpected error retrieving secret {secret_name}: {str(e)}")