fix(parameters): handle transform failures gracefully in get_parameters_by_name

aws_lambda_powertools/utilities/parameters/ssm.py

@@ -27,7 +27,11 @@
     SSM_PARAMETER_TIER,
     SSM_PARAMETER_TYPES,
 )
-from aws_lambda_powertools.utilities.parameters.exceptions import GetParameterError, SetParameterError
+from aws_lambda_powertools.utilities.parameters.exceptions import (
+    GetParameterError,
+    SetParameterError,
+    TransformParameterError,
+)
 from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning
 
 if TYPE_CHECKING:
@@ -532,7 +536,7 @@ def _get_parameters_by_name_with_decrypt_option(
         for parameter, options in batch.items():
             try:
                 response[parameter] = self.get(parameter, options["max_age"], options["transform"], options["decrypt"])
-            except GetParameterError:
+            except (GetParameterError, TransformParameterError):
                 if raise_on_error:
                     raise
                 errors.append(parameter)
@@ -634,7 +638,12 @@ def _get_parameters_by_name(
             return ret, parameter_names
 
         batch_errors = self._handle_any_invalid_get_parameter_errors(response, raise_on_error)
-        transformed_params = self._transform_and_cache_get_parameters_response(response, parameters, raise_on_error)
+        transformed_params, transform_errors = self._transform_and_cache_get_parameters_response(
+            response,
+            parameters,
+            raise_on_error,
+        )
+        batch_errors.extend(transform_errors)
 
         return transformed_params, batch_errors
 
@@ -643,8 +652,9 @@ def _transform_and_cache_get_parameters_response(
         api_response: GetParametersResultTypeDef,
         parameters: dict[str, Any],
         raise_on_error: bool = True,
-    ) -> dict[str, Any]:
+    ) -> tuple[dict[str, Any], list[str]]:
         response: dict[str, Any] = {}
+        transform_errors: list[str] = []
 
         for parameter in api_response["Parameters"]:
             name = parameter["Name"]
@@ -654,19 +664,28 @@ def _transform_and_cache_get_parameters_response(
 
             # NOTE: If transform is set, we do it before caching to reduce number of operations
             if transform:
-                value = transform_value(
-                    value=value,
-                    transform=transform,
-                    raise_on_transform_error=raise_on_error,
-                    key=name,
-                )  # type: ignore[assignment]
+                try:
+                    value = transform_value(
+                        value=value,
+                        transform=transform,
+                        raise_on_transform_error=True,
+                        key=name,
+                    )  # type: ignore[assignment]
+                except TransformParameterError:
+                    if raise_on_error:
+                        raise
+
+                    # NOTE: Don't cache failed transforms, otherwise a corrected parameter
+                    # would keep returning None from the cache until max_age expires
+                    transform_errors.append(name)
+                    continue
 
             _cache_key = (name, transform)
             self.add_to_cache(key=_cache_key, value=value, max_age=options["max_age"])
 
             response[name] = value
 
-        return response
+        return response, transform_errors
 
     @staticmethod
     def _handle_any_invalid_get_parameter_errors(

tests/functional/parameters/_boto3/test_utilities_parameters.py

@@ -1142,6 +1142,146 @@ def test_ssm_provider_get_parameters_by_name_do_not_raise_on_failure_transform(m
         stubber.deactivate()
 
 
+def test_ssm_provider_get_parameters_by_name_do_not_raise_on_transform_failure_with_decrypt(mock_name):
+    # GIVEN one parameter needs decryption and its value can't be transformed
+    # and a second parameter is successfully fetched through the batch API
+    decrypt_fail = f"/decrypt/{mock_name}"
+    success = f"/batch/{mock_name}"
+    params = {decrypt_fail: {"decrypt": True}, success: {}}
+    expected_value = {"value": "success"}
+
+    class FakeClient:
+        def get_parameter(self, **kwargs):
+            assert kwargs["Name"] == decrypt_fail
+            assert kwargs["WithDecryption"] is True
+            return {"Parameter": {"Name": decrypt_fail, "Value": "not json", "Type": "String", "Version": 1}}
+
+        def get_parameters(self, **kwargs):
+            assert kwargs["Names"] == [success]
+            return build_get_parameters_stub(params={success: json.dumps(expected_value)})
+
+    provider = parameters.SSMProvider(boto3_client=FakeClient())
+
+    # WHEN fail-fast is disabled in get_parameters_by_name
+    ret = provider.get_parameters_by_name(parameters=params, transform="json", raise_on_error=False)
+
+    # THEN no error is raised, the transform failure is reported under "_errors",
+    # and its value is not returned
+    assert ret[success] == expected_value
+    assert ret["_errors"] == [decrypt_fail]
+    assert decrypt_fail not in ret
+
+
+def test_ssm_provider_get_parameters_by_name_do_not_cache_failed_transform(mock_name):
+    # GIVEN a parameter whose value can't be transformed
+    param = f"/{mock_name}"
+    params = {param: {}}
+    corrected_value = {"value": "corrected"}
+
+    class FakeClient:
+        def __init__(self):
+            self.get_parameters_calls = 0
+
+        def get_parameters(self, **kwargs):
+            self.get_parameters_calls += 1
+            assert kwargs["Names"] == [param]
+
+            if self.get_parameters_calls == 1:
+                return build_get_parameters_stub(params={param: "not json"})
+
+            return build_get_parameters_stub(params={param: json.dumps(corrected_value)})
+
+    client = FakeClient()
+    provider = parameters.SSMProvider(boto3_client=client)
+
+    # WHEN fail-fast is disabled in get_parameters_by_name
+    ret = provider.get_parameters_by_name(parameters=params, transform="json", raise_on_error=False)
+
+    # THEN the transform failure is reported and not cached
+    assert ret["_errors"] == [param]
+    assert param not in ret
+    assert (param, "json") not in provider.store
+
+    # WHEN fetching again after the parameter is corrected
+    ret = provider.get_parameters_by_name(parameters=params, transform="json", raise_on_error=False)
+
+    # THEN a fresh request is made and the corrected value is returned instead of a cached failure
+    assert client.get_parameters_calls == 2
+    assert ret["_errors"] == []
+    assert ret[param] == corrected_value
+    assert (param, "json") in provider.store
+
+
+def test_ssm_provider_get_parameters_by_name_raise_on_transform_failure_with_decrypt(mock_name, config):
+    # GIVEN one parameter needs decryption and its value can't be transformed
+    decrypt_fail = f"/decrypt/{mock_name}"
+    success = f"/batch/{mock_name}"
+    params = {decrypt_fail: {"decrypt": True}, success: {}}
+
+    provider = parameters.SSMProvider(boto_config=config)
+    stubber = stub.Stubber(provider.client)
+    stubber.add_response(
+        "get_parameter",
+        {"Parameter": {"Name": decrypt_fail, "Value": "not json", "Type": "String", "Version": 1}},
+        {"Name": decrypt_fail, "WithDecryption": True},
+    )
+    stubber.activate()
+
+    # WHEN using get_parameters_by_name with fail-fast enabled (default)
+    try:
+        with pytest.raises(parameters.TransformParameterError):
+            provider.get_parameters_by_name(parameters=params, transform="json")
+
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
+def test_ssm_provider_get_parameters_by_name_raise_on_batch_transform_failure(mock_name):
+    # GIVEN a batch parameter whose value can't be transformed
+    param = f"/{mock_name}"
+    params = {param: {}}
+
+    class FakeClient:
+        def get_parameters(self, **kwargs):
+            assert kwargs["Names"] == [param]
+            return build_get_parameters_stub(params={param: "not json"})
+
+    provider = parameters.SSMProvider(boto3_client=FakeClient())
+
+    # WHEN using get_parameters_by_name with fail-fast enabled (default)
+    with pytest.raises(parameters.TransformParameterError):
+        provider.get_parameters_by_name(parameters=params, transform="json")
+
+
+def test_ssm_provider_get_parameters_by_name_json_null_is_not_a_transform_failure(mock_name):
+    # GIVEN parameters whose values are valid JSON null
+    decrypt_param = f"/decrypt/{mock_name}"
+    batch_param = f"/batch/{mock_name}"
+    params = {decrypt_param: {"decrypt": True}, batch_param: {}}
+
+    class FakeClient:
+        def get_parameter(self, **kwargs):
+            assert kwargs["Name"] == decrypt_param
+            assert kwargs["WithDecryption"] is True
+            return {"Parameter": {"Name": decrypt_param, "Value": "null", "Type": "String", "Version": 1}}
+
+        def get_parameters(self, **kwargs):
+            assert kwargs["Names"] == [batch_param]
+            return build_get_parameters_stub(params={batch_param: "null"})
+
+    provider = parameters.SSMProvider(boto3_client=FakeClient())
+
+    # WHEN fail-fast is disabled in get_parameters_by_name
+    ret = provider.get_parameters_by_name(parameters=params, transform="json", raise_on_error=False)
+
+    # THEN JSON null is a successful transform, not reported as a failure
+    assert ret["_errors"] == []
+    assert ret[decrypt_param] is None
+    assert ret[batch_param] is None
+    assert (batch_param, "json") in provider.store
+
+
 def test_ssm_provider_get_parameters_by_name_auto_transform_infers_from_parameter_name(mock_name, mock_value, config):
     # GIVEN transform="auto" and a parameter name suffixed with ".json"
     # (auto-transform must infer the transform method from the parameter name, not its value)