import logging from botocore.exceptions import ClientError import boto3 import config logger = logging.getLogger(__name__) def get_secret(secret): secret = '' secret_name = config.DD_SECRET_NAME endpoint_url = 'https://secretsmanager.us-east-1.amazonaws.com' session = boto3.session.Session() client = session.client( service_name='secretsmanager', region_name='us-east-1', endpoint_url=endpoint_url ) try: get_secret_value_response = client.get_secret_value( SecretId=secret_name ) logger.info('Retrieved secret from secrets manager') except ClientError as e: if e.response['Error']['Code'] == 'ResourceNotFoundException': logger.info( 'The requested secret ' + secret_name + ' was not found') elif e.response['Error']['Code'] == 'InvalidRequestException': logger.info('The request was invalid due to:', e) elif e.response['Error']['Code'] == 'InvalidParameterException': logger.info('The request had invalid params:', e) elif 'Error' in e.response: logger.info('Error:', e) else: # Decrypted secret using the associated KMS CMK # Depending on whether the secret was a string or binary, # one of these fields will be populated if 'SecretString' in get_secret_value_response: secret = get_secret_value_response['SecretString'] else: secret = get_secret_value_response['SecretBinary'] return secret