feat(parameters): transform = "auto"

Add a new tranform type called "auto", which looks at the
key name to autodetect the encoding of the values in dynamodb or
the parameter store.

Keys ending with ".json" are assumed to be "json" string
Keys ending with ".binary" are assumed to be "binary" base64 decode
strings

Author: michaelbrewer

aws_lambda_powertools/utilities/parameters/base.py

@@ -115,8 +115,8 @@ def get_multiple(
             Maximum age of the cached value
         transform: str, optional
             Optional transformation of the parameter value. Supported values
-            are "json" for JSON strings and "binary" for base 64 encoded
-            values.
+            are "json" for JSON strings, "binary" for base 64 encoded
+            values or "auto" which looks at the attribute key to determine the type.
         raise_on_transform_error: bool, optional
             Raises an exception if any transform fails, otherwise this will
             return a None value for each transform that failed
@@ -145,7 +145,11 @@ def get_multiple(
 
         if transform is not None:
             for (key, value) in values.items():
-                values[key] = transform_value(value, transform, raise_on_transform_error)
+                _transform = get_transform_method(key, transform)
+                if _transform is None:
+                    continue
+
+                values[key] = transform_value(value, _transform, raise_on_transform_error)
 
         self.store[key] = ExpirableValue(values, datetime.now() + timedelta(seconds=max_age),)
 
@@ -159,6 +163,18 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
         raise NotImplementedError()
 
 
+def get_transform_method(key: str, transform: str) -> Union[str, None]:
+    if transform != "auto":
+        return transform
+
+    if key.endswith(".json"):
+        return "json"
+    elif key.endswith(".binary"):
+        return "binary"
+    else:
+        return None
+
+
 def transform_value(value: str, transform: str, raise_on_transform_error: bool = True) -> Union[dict, bytes, None]:
     """
     Apply a transform to a value

tests/functional/test_utilities_parameters.py

@@ -233,6 +233,48 @@ def test_dynamodb_provider_get_multiple(mock_name, mock_value, config):
         stubber.deactivate()
 
 
+def test_dynamodb_provider_get_multiple_auto(mock_name, mock_value, config):
+    """
+    Test DynamoDBProvider.get_multiple() with transform = "auto"
+    """
+    mock_binary = mock_value.encode()
+    mock_binary_data = base64.b64encode(mock_binary).decode()
+    mock_json_data = json.dumps({mock_name: mock_value})
+    mock_params = {"D.json": mock_json_data, "E.binary": mock_binary_data, "F": mock_value}
+    table_name = "TEST_TABLE_AUTO"
+
+    # Create a new provider
+    provider = parameters.DynamoDBProvider(table_name, config=config)
+
+    # Stub the boto3 client
+    stubber = stub.Stubber(provider.table.meta.client)
+    response = {
+        "Items": [
+            {"id": {"S": mock_name}, "sk": {"S": name}, "value": {"S": value}} for (name, value) in mock_params.items()
+        ]
+    }
+    expected_params = {"TableName": table_name, "KeyConditionExpression": Key("id").eq(mock_name)}
+    stubber.add_response("query", response, expected_params)
+    stubber.activate()
+
+    try:
+        values = provider.get_multiple(mock_name, transform="auto")
+
+        stubber.assert_no_pending_responses()
+
+        assert len(values) == len(mock_params)
+        for key in mock_params.keys():
+            assert key in values
+            if key.endswith(".json"):
+                assert values[key][mock_name] == mock_value
+            elif key.endswith(".binary"):
+                assert values[key] == mock_binary
+            else:
+                assert values[key] == mock_value
+    finally:
+        stubber.deactivate()
+
+
 def test_dynamodb_provider_get_multiple_next_token(mock_name, mock_value, config):
     """
     Test DynamoDBProvider.get_multiple() with a non-cached path