import json import logging import boto3 from botocore.exceptions import ClientError log = logging.getLogger(__name__) class SecretsManagerClient: """Client to get test credentials from AWS Secrets Manager.""" def __init__(self, region_name: str = 'us-east-1'): self.client = boto3.client('secretsmanager', region_name=region_name) def get_secret(self, secret_name: str) -> str: """Get a secret string value.""" log.info(f'Getting secret from AWS Secrets Manager: {secret_name}') try: response = self.client.get_secret_value(SecretId=secret_name) except ClientError: log.exception(f"Failed to get secret '{secret_name}'") raise return response['SecretString'] def get_secret_json(self, secret_name: str) -> dict: """Get a secret as JSON object.""" log.info(f'Getting JSON secret: {secret_name}') secret_string = self.get_secret(secret_name) return json.loads(secret_string) def put_secret_json(self, secret_name: str, secret_value: dict) -> None: """Update a secret with JSON value.""" log.info(f'Updating JSON secret: {secret_name}') try: self.client.put_secret_value( SecretId=secret_name, SecretString=json.dumps(secret_value, indent=2) ) except ClientError: log.exception(f"Failed to update secret '{secret_name}'") raise