| | 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: |
| | |
| | secret_value = response["SecretString"] |
| | else: |
| | |
| | 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)}") |
| |
|