File size: 2,228 Bytes
a103028 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 |
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)}")
|