#12 Use PEP-8 formatting

src/value_fetcher/__init__.py

@@ -1,4 +1,7 @@
 """
 Load value fetcher class to make available at the package level
 """
+
 from .base import ValueFetcher
+
+type(ValueFetcher)

src/value_fetcher/aws.py

@@ -8,14 +8,16 @@
 import boto3
 import botocore
 
+
 class Aws:
     """
     Fetch parameters and send emails.
     """
+
     def __init__(self) -> None:
         # Prepare AWS clients
-        self.ssm = boto3.client('ssm')
-        self.secretsmanager = boto3.client('secretsmanager')
+        self.ssm = boto3.client("ssm")
+        self.secretsmanager = boto3.client("secretsmanager")
 
     def get_parameter_value(self, name) -> str:
         """
@@ -25,15 +27,12 @@ def get_parameter_value(self, name) -> str:
 
         value = None
 
-        logging.debug('Fetching AWS SSM Parameter Store parameter: %s', name)
+        logging.debug("Fetching AWS SSM Parameter Store parameter: %s", name)
         try:
-            response = self.ssm.get_parameter(
-                Name=name,
-                WithDecryption=True
-            )
-            if 'Parameter' in response:
-                if 'Value' in response['Parameter']:
-                    value = response['Parameter']['Value']
+            response = self.ssm.get_parameter(Name=name, WithDecryption=True)
+            if "Parameter" in response:
+                if "Value" in response["Parameter"]:
+                    value = response["Parameter"]["Value"]
         except (
             botocore.exceptions.ClientError,
             botocore.exceptions.NoCredentialsError,
@@ -42,29 +41,28 @@ def get_parameter_value(self, name) -> str:
             self.ssm.exceptions.ParameterNotFound,
             self.ssm.exceptions.ParameterVersionNotFound,
         ) as exception:
-            logging.warning('Unable to retrieve AWS SSM Parameter value for name %s', name)
+            logging.warning("AWS SSM Parameter fetch failed: %s", name)
             logging.warning(exception)
 
         return value
 
-
     def get_secret_value(self, name) -> str:
         """
         Fetch encrypted secret from Secrets Manager
         https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html
         """
 
         value = None
-        logging.debug('Fetching AWS Secrets Manager secret: %s', name)
+        logging.debug("Fetching AWS Secrets Manager secret: %s", name)
 
         try:
             response = self.secretsmanager.get_secret_value(SecretId=name)
-            if 'SecretString' in response:
-                value = response['SecretString']
-            elif 'SecretBinary' in response:
-                value = response['SecretBinary']
+            if "SecretString" in response:
+                value = response["SecretString"]
+            elif "SecretBinary" in response:
+                value = response["SecretBinary"]
                 if isinstance(value, bytes):
-                    value = value.decode('utf-8')
+                    value = value.decode("utf-8")
         except (
             botocore.exceptions.ClientError,
             botocore.exceptions.NoCredentialsError,
@@ -74,7 +72,7 @@ def get_secret_value(self, name) -> str:
             self.secretsmanager.exceptions.DecryptionFailure,
             self.secretsmanager.exceptions.InternalServiceError,
         ) as exception:
-            logging.warning('Unable to retrieve AWS Secrets Manager value for name %s', name)
+            logging.warning("AWS Secrets Manager fetch failed: %s", name)
             logging.warning(exception)
 
         return value

src/value_fetcher/base.py

@@ -14,6 +14,7 @@ class ValueFetcher:
     """
     Configuration provider class
     """
+
     def __init__(self, env_defaults: dict = None) -> None:
         self.env_defaults = env_defaults
 
@@ -26,115 +27,115 @@ def get(self, key: str) -> str:
         key = key.upper()
 
         if not isinstance(key, str) or len(key) == 0:
-            message = 'Invalid key name provided'
+            message = "Invalid key name provided"
             logging.critical(message)
             logging.critical(key)
             raise ValueError(message)
 
-        logging.debug('Retrieving value for key %s', key)
-        source = os.environ.get(f'{key}_SOURCE', 'env').lower()
-        logging.debug('%s source: %s', key, source)
+        logging.debug("Retrieving value for key %s", key)
+        source = os.environ.get(f"{key}_SOURCE", "env").lower()
+        logging.debug("%s source: %s", key, source)
         # Get value from environment variable
-        if source == 'env':
+        if source == "env":
             return self.get_from_env(key)
 
         # Get value from AWS SSM parameter store
-        if source == 'aws_ssm_parameter_store':
+        if source == "aws_ssm_parameter_store":
             return self.get_from_aws_ssm_parameter_store(key)
 
         # Get value from AWS Secrets Manager
-        if source == 'aws_secrets_manager':
+        if source == "aws_secrets_manager":
             return self.get_from_aws_secrets_manager(key)
 
-        message = f'Unknown source {source}'
+        message = f"Unknown source {source}"
         logging.critical(message)
         raise ValueError(message)
 
-
     def get_from_env(self, key) -> str:
         """
         Fetch a value from environment variables, using a default if available.
         If value is empty and cannot be found in defaults, raise exception.
         """
 
         if not isinstance(key, str) or len(key) == 0:
-            message = 'Missing or empty environment key'
+            message = "Missing or empty environment key"
             logging.critical(message)
             logging.critical(key)
             raise ValueError(message)
 
-        logging.debug('Checking environment for key: %s', key)
-        value = os.environ.get(key, '').strip()
+        logging.debug("Checking environment for key: %s", key)
+        value = os.environ.get(key, "").strip()
         if not isinstance(value, str) or len(value) == 0:
-            if isinstance(self.env_defaults, dict) and key in self.env_defaults:
-                logging.debug('Using default value for environment variable: %s', key)
-                value = self.env_defaults[key]
+            defaults = self.env_defaults
+            if isinstance(defaults, dict) and key in defaults:
+                logging.debug("Using default value for env var: %s", key)
+                value = defaults[key]
             else:
-                message = f'Missing or empty environment value for {key}'
+                message = f"Missing or empty environment value for {key}"
                 logging.critical(message)
                 raise ValueError(message)
 
         return value
 
-
     def get_from_aws_ssm_parameter_store(self, key: str) -> str:
         """
         Fetch a value from AWS SSM Parameter store.
-        If the parameter name is not configured by an environment variable, try the provided key.
+        If the parameter name is not configured by an environment variable,
+        try the provided key.
         """
 
         if not isinstance(key, str) or len(key) == 0:
-            message = 'Missing or empty AWS SSM Parameter Store key'
+            message = "Missing or empty AWS SSM Parameter Store key"
             logging.critical(message)
             raise ValueError(message)
 
-        name_key = f'{key}_PARAMETER_STORE_NAME'
+        name_key = f"{key}_PARAMETER_STORE_NAME"
 
-        logging.debug('Checking environment for AWS SSM Parameter name: %s', name_key)
+        logging.debug("Checking env for AWS SSM Parameter %s", name_key)
         try:
             name = self.get_from_env(name_key)
-            logging.debug('Using parameter name %s', name)
+            logging.debug("Using parameter name %s", name)
         except ValueError as exception:
             logging.debug(exception)
-            logging.debug('Using key name for parameter %s', key)
+            logging.debug("Using key name for parameter %s", key)
             name = key
 
         aws = Aws()
         value = aws.get_parameter_value(name)
         if not isinstance(value, str) or len(value) == 0:
-            message = f'Missing or empty AWS SSM Parameter Store value for {name}'
+            message = f"Missing or empty AWS SSM Parameter Store {name}"
             logging.critical(message)
             raise ValueError(message)
 
         return value
 
-
     def get_from_aws_secrets_manager(self, key: str) -> str:
         """
         Fetch a value from AWS Secrets Manager.
-        If the secret name is not configured by an environment variable, use the key directly.
+        If the secret name is not configured by an environment variable,
+        use the key directly.
         """
 
         if not isinstance(key, str) or len(key) == 0:
-            message = 'Missing or empty AWS Secrets Manager key'
+            message = "Missing or empty AWS Secrets Manager key"
             logging.critical(message)
             raise ValueError(message)
 
-        name_key = f'{key}_SECRETS_MANAGER_NAME'
+        name_key = f"{key}_SECRETS_MANAGER_NAME"
 
-        logging.debug('Checking environment for AWS Secret Manager name: %s', name_key)
+        logging.debug("Checking env for AWS Secret Manager %s", name_key)
         try:
             name = self.get_from_env(name_key)
-            logging.debug('Using secret name %s', name)
+            logging.debug("Using secret name %s", name)
         except ValueError as exception:
             logging.debug(exception)
-            logging.debug('Using key name for secret %s', key)
+            logging.debug("Using key name for secret %s", key)
             name = key
 
         aws = Aws()
         value = aws.get_secret_value(name)
         if not isinstance(value, str) or len(value) == 0:
-            message = f'Missing or empty AWS Secrets Manager value for {name}'
+            message = f"Missing or empty AWS Secrets Manager value for {name}"
             logging.critical(message)
             raise ValueError(message)
 

tests/test_aws.py

@@ -3,14 +3,15 @@
 """
 
 import os
-from moto import mock_ssm, mock_secretsmanager
+from moto import mock_aws
 from value_fetcher.aws import Aws
 from tests import utils
 
 # Set boto client default values
-os.environ['AWS_DEFAULT_REGION'] = 'eu-west-2'
+os.environ["AWS_DEFAULT_REGION"] = "eu-west-2"
 
-@mock_ssm
+
+@mock_aws
 def test_getting_parameter():
     """
     Check an SSM parameter is returned
@@ -21,27 +22,27 @@ def test_getting_parameter():
 
     # Fetch and verify parameter
     aws = Aws()
-    param = aws.get_parameter_value('test')
-    assert param == 'abc'
+    param = aws.get_parameter_value("test")
+    assert param == "abc"
 
     # Assert missing parameter returns nothing
-    assert not aws.get_parameter_value('doesnotexist')
+    assert not aws.get_parameter_value("doesnotexist")
 
 
-@mock_secretsmanager
+@mock_aws
 def test_getting_secret():
     """
     Check string and binary secrets may be retrieved
     """
 
     # Create mocked secrets to later fetch using moto
-    secret_string = 'my secret value'
-    secret_binary = bytes(secret_string, 'utf-8')
-    utils.put_secretsmanager_secret('/secret/string', secret_string)
-    utils.put_secretsmanager_secret('/secret/binary', None, secret_binary)
+    secret_string = "my secret value"
+    secret_binary = bytes(secret_string, "utf-8")
+    utils.put_secretsmanager_secret("/secret/string", secret_string)
+    utils.put_secretsmanager_secret("/secret/binary", None, secret_binary)
 
     # Fetch secrets
     aws = Aws()
-    assert aws.get_secret_value('/secret/string') == secret_string
-    assert aws.get_secret_value('/secret/binary') == secret_string
-    assert not aws.get_secret_value('/does/not/exist')
+    assert aws.get_secret_value("/secret/string") == secret_string
+    assert aws.get_secret_value("/secret/binary") == secret_string
+    assert not aws.get_secret_value("/does/not/exist")

tests/test_base.py

@@ -8,33 +8,34 @@
 from value_fetcher import ValueFetcher
 
 # Set mocked boto (moto) client default values
-os.environ['AWS_DEFAULT_REGION'] = 'eu-west-2'
+os.environ["AWS_DEFAULT_REGION"] = "eu-west-2"
 
 fetcher = ValueFetcher()
 
+
 def test_invalid_names():
     """
     Assert invalid or empty names produce exceptions
     """
     # Test non-existent variable throws an exception
     with pytest.raises(ValueError) as exception:
-        fetcher.get('')
+        fetcher.get("")
 
     with pytest.raises(ValueError) as env_exception:
-        fetcher.get_from_env('')
+        fetcher.get_from_env("")
 
-    assert str(exception.value) == 'Invalid key name provided'
-    assert str(env_exception.value) == 'Missing or empty environment key'
+    assert str(exception.value) == "Invalid key name provided"
+    assert str(env_exception.value) == "Missing or empty environment key"
 
 
 def test_unknown_source(monkeypatch):
     """
     Test unknown source raises an exception
     """
-    monkeypatch.setenv('UNKNOWN', 'test_value')
-    monkeypatch.setenv('UNKNOWN_SOURCE', 'unknown')
+    monkeypatch.setenv("UNKNOWN", "test_value")
+    monkeypatch.setenv("UNKNOWN_SOURCE", "unknown")
     # Test unknown source returns nothing
     with pytest.raises(ValueError) as exception:
-        fetcher.get('UNKNOWN')
+        fetcher.get("UNKNOWN")
 
-    assert 'Unknown source' in str(exception.value)
+    assert "Unknown source" in str(exception.value)