|
|
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)}") |
|
|
|
