-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #32 from NYPL/add-secrets-manager-client
Add SecretsManager client
- Loading branch information
Showing
7 changed files
with
136 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,67 @@ | ||
import boto3 | ||
import json | ||
import os | ||
|
||
from botocore.exceptions import ClientError | ||
from nypl_py_utils.functions.log_helper import create_log | ||
|
||
|
||
class SecretsManagerClient: | ||
"""Client for interacting with AWS Secrets Manager""" | ||
|
||
def __init__(self): | ||
self.logger = create_log('secrets_manager_client') | ||
|
||
try: | ||
self.secrets_manager_client = boto3.client( | ||
'secretsmanager', region_name=os.environ.get('AWS_REGION', | ||
'us-east-1')) | ||
except ClientError as e: | ||
self.logger.error( | ||
'Could not create Secrets Manager client: {err}'.format( | ||
err=e)) | ||
raise SecretsManagerClientError( | ||
'Could not create Secrets Manager client: {err}'.format( | ||
err=e)) from None | ||
|
||
def close(self): | ||
self.secrets_manager_client.close() | ||
|
||
def get_secret(self, secret_name, is_json=True): | ||
""" | ||
Retrieves secret with the given name from the Secrets Manager. | ||
Parameters | ||
---------- | ||
secret_name: str | ||
The name of the secret to retrieve | ||
is_json: bool, optional | ||
Whether the value of the secret is a JSON string that should be | ||
returned as a dictionary | ||
Returns | ||
------- | ||
dict or str | ||
Dictionary if `is_json` is True; string if `is_json` is False | ||
""" | ||
self.logger.debug('Retrieving \'{}\' from Secrets Manager'.format( | ||
secret_name)) | ||
try: | ||
response = self.secrets_manager_client.get_secret_value( | ||
SecretId=secret_name) | ||
if is_json: | ||
return json.loads(response['SecretString']) | ||
else: | ||
return response['SecretString'] | ||
except ClientError as e: | ||
self.logger.error( | ||
('Could not retrieve \'{secret}\' from Secrets Manager: {err}') | ||
.format(secret=secret_name, err=e)) | ||
raise SecretsManagerClientError( | ||
('Could not retrieve \'{secret}\' from Secrets Manager: {err}') | ||
.format(secret=secret_name, err=e)) from None | ||
|
||
|
||
class SecretsManagerClientError(Exception): | ||
def __init__(self, message=None): | ||
self.message = message |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,55 @@ | ||
import pytest | ||
|
||
from botocore.exceptions import ClientError | ||
from datetime import datetime | ||
from nypl_py_utils.classes.secrets_manager_client import ( | ||
SecretsManagerClient, SecretsManagerClientError) | ||
|
||
_TEST_RESPONSE = { | ||
'ARN': 'test_arn', | ||
'Name': 'test_secret', | ||
'VersionId': 'test_version', | ||
'SecretString': '{\n "key1": "value1",\n "key2": "value2"\n}', | ||
'VersionStages': ['AWSCURRENT'], | ||
'CreatedDate': datetime(2024, 1, 1, 1, 1, 1, 1), | ||
'ResponseMetadata': { | ||
'RequestId': 'test-request-id', | ||
'HTTPStatusCode': 200, | ||
'HTTPHeaders': { | ||
'x-amzn-requestid': 'test-request-id', | ||
'content-type': 'application/x-amz-json-1.1', | ||
'content-length': '155', | ||
'date': 'Mon, 1 Jan 2024 07:01:01 GMT' | ||
}, | ||
'RetryAttempts': 0} | ||
} | ||
|
||
|
||
class TestSecretsManagerClient: | ||
|
||
@pytest.fixture | ||
def test_instance(self, mocker): | ||
mocker.patch('boto3.client') | ||
return SecretsManagerClient() | ||
|
||
def test_get_secret(self, test_instance): | ||
test_instance.secrets_manager_client.get_secret_value.return_value = \ | ||
_TEST_RESPONSE | ||
assert test_instance.get_secret('test_secret') == { | ||
'key1': 'value1', 'key2': 'value2'} | ||
test_instance.secrets_manager_client.get_secret_value\ | ||
.assert_called_once_with(SecretId='test_secret') | ||
|
||
def test_get_secret_non_json(self, test_instance): | ||
test_instance.secrets_manager_client.get_secret_value.return_value = \ | ||
_TEST_RESPONSE | ||
assert test_instance.get_secret('test_secret', is_json=False) == ( | ||
'{\n "key1": "value1",\n "key2": "value2"\n}') | ||
test_instance.secrets_manager_client.get_secret_value\ | ||
.assert_called_once_with(SecretId='test_secret') | ||
|
||
def test_get_secret_error(self, test_instance): | ||
test_instance.secrets_manager_client.get_secret_value.side_effect = \ | ||
ClientError({}, 'GetSecretValue') | ||
with pytest.raises(SecretsManagerClientError): | ||
test_instance.get_secret('test_secret') |