feat(event_sources): Add __str__ to Data Classes base DictWrapper

aws_lambda_powertools/utilities/data_classes/code_pipeline_job_event.py

@@ -80,6 +80,8 @@ def location(self) -> CodePipelineLocation:
 
 
 class CodePipelineArtifactCredentials(DictWrapper):
+    _sensitive_properties = ["secret_access_key", "session_token"]
+
     @property
     def access_key_id(self) -> str:
         return self["accessKeyId"]

aws_lambda_powertools/utilities/data_classes/common.py

@@ -1,7 +1,7 @@
 import base64
 import json
 from collections.abc import Mapping
-from typing import Any, Dict, Iterator, Optional
+from typing import Any, Dict, Iterator, List, Optional
 
 from aws_lambda_powertools.shared.headers_serializer import BaseHeadersSerializer
 
@@ -28,6 +28,49 @@ def __iter__(self) -> Iterator:
     def __len__(self) -> int:
         return len(self._data)
 
+    def __str__(self) -> str:
+        return str(self._str_helper())
+
+    def _str_helper(self) -> Dict[str, Any]:
+        """
+        Recursively get a Dictionary of DictWrapper properties primarily
+        for use by __str__ for debugging purposes.
+
+        Will remove "raw_event" properties, and any defined by the Data Class
+        `_sensitive_properties` list field.
+        This should be used in case where secrets, such as access keys, are
+        stored in the Data Class but should not be logged out.
+        """
+        properties = self._properties()
+        sensitive_properties = ["raw_event"]
+        if hasattr(self, "_sensitive_properties"):
+            sensitive_properties.extend(self._sensitive_properties)  # pyright: ignore
+
+        result: Dict[str, Any] = {}
+        for property_key in properties:
+            if property_key in sensitive_properties:
+                result[property_key] = "[SENSITIVE]"
+            else:
+                try:
+                    property_value = getattr(self, property_key)
+                    result[property_key] = property_value
+
+                    # Checks whether the class is a subclass of the parent class to perform a recursive operation.
+                    if issubclass(property_value.__class__, DictWrapper):
+                        result[property_key] = property_value._str_helper()
+                    # Checks if the key is a list and if it is a subclass of the parent class
+                    elif isinstance(property_value, list):
+                        for seq, item in enumerate(property_value):
+                            if issubclass(item.__class__, DictWrapper):
+                                result[property_key][seq] = item._str_helper()
+                except Exception:
+                    result[property_key] = "[Cannot be deserialized]"
+
+        return result
+
+    def _properties(self) -> List[str]:
+        return [p for p in dir(self.__class__) if isinstance(getattr(self.__class__, p), property)]
+
     def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
         return self._data.get(key, default)
 

docs/utilities/data_classes.md

@@ -52,6 +52,22 @@ Same example as above, but using the `event_source` decorator
         if 'helloworld' in event.path and event.http_method == 'GET':
             do_something_with(event.body, user)
     ```
+
+Log Data Event for Troubleshooting
+
+=== "app.py"
+
+    ```python hl_lines="4 8"
+    from aws_lambda_powertools.utilities.data_classes import event_source, APIGatewayProxyEvent
+    from aws_lambda_powertools.logging.logger import Logger
+
+    logger = Logger(service="hello_logs", level="DEBUG")
+
+    @event_source(data_class=APIGatewayProxyEvent)
+    def lambda_handler(event: APIGatewayProxyEvent, context):
+        logger.debug(event)
+    ```
+
 **Autocomplete with self-documented properties and methods**
 
 ![Utilities Data Classes](../media/utilities_data_classes.png)
@@ -1104,3 +1120,28 @@ This example is based on the AWS Blog post [Introducing Amazon S3 Object Lambda
         for record in event.records:
             do_something_with(record.body)
     ```
+
+## Advanced
+
+### Debugging
+
+Alternatively, you can print out the fields to obtain more information. All classes come with a `__str__` method that generates a dictionary string which can be quite useful for debugging.
+
+However, certain events may contain sensitive fields such as `secret_access_key` and `session_token`, which are labeled as `[SENSITIVE]` to prevent any accidental disclosure of confidential information.
+
+!!! warning "Some fields contain user-supplied data, which can be plain text or JSON. If deserialization of this field fails, it will appear as [Cannot be deserialized]"
+
+=== "debugging.py"
+    ```python hl_lines="9"
+    --8<-- "examples/event_sources/src/debugging.py"
+    ```
+
+=== "debugging_event.json"
+    ```json hl_lines="28 29"
+    --8<-- "examples/event_sources/src/debugging_event.json"
+    ```
+=== "debugging_output.json"
+    ```json hl_lines="16 17 18"
+    --8<-- "examples/event_sources/src/debugging_output.json"
+    ```
+    ```

examples/event_sources/src/debugging.py

@@ -0,0 +1,9 @@
+from aws_lambda_powertools.utilities.data_classes import (
+    CodePipelineJobEvent,
+    event_source,
+)
+
+
+@event_source(data_class=CodePipelineJobEvent)
+def lambda_handler(event, context):
+    print(event)

examples/event_sources/src/debugging_event.json

@@ -0,0 +1,34 @@
+{
+    "CodePipeline.job": {
+        "id": "11111111-abcd-1111-abcd-111111abcdef",
+        "accountId": "111111111111",
+        "data": {
+            "actionConfiguration": {
+                "configuration": {
+                    "FunctionName": "MyLambdaFunctionForAWSCodePipeline",
+                    "UserParameters": "some-input-such-as-a-URL"
+                }
+            },
+            "inputArtifacts": [
+                {
+                    "name": "ArtifactName",
+                    "revision": null,
+                    "location": {
+                        "type": "S3",
+                        "s3Location": {
+                            "bucketName": "the name of the bucket configured as the pipeline artifact store in Amazon S3, for example codepipeline-us-east-2-1234567890",
+                            "objectKey": "the name of the application, for example CodePipelineDemoApplication.zip"
+                        }
+                    }
+                }
+            ],
+            "outputArtifacts": [],
+            "artifactCredentials": {
+                "accessKeyId": "AKIAIOSFODNN7EXAMPLE",
+                "secretAccessKey": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+                "sessionToken": "MIICiTCCAfICCQD6m7oRw0uXOjANBgkqhkiG9w0BAQUFADCBiDELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAldBMRAwDgYDVQQHEwdTZWF0dGxlMQ8wDQYDVQQKEwZBbWF6b24xFDASBgNVBAsTC0lBTSBDb25zb2xlMRIwEAYDVQQDEwlUZXN0Q2lsYWMxHzAdBgkqhkiG9w0BCQEWEG5vb25lQGFtYXpvbi5jb20wHhcNMTEwNDI1MjA0NTIxWhcNMTIwNDI0MjA0NTIxWjCBiDELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAldBMRAwDgYDVQQHEwdTZWF0dGxlMQ8wDQYDVQQKEwZBbWF6b24xFDASBgNVBAsTC0lBTSBDb25zb2xlMRIwEAYDVQQDEwlUZXN0Q2lsYWMxHzAdBgkqhkiG9w0BCQEWEG5vb25lQGFtYXpvbi5jb20wgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBAMaK0dn+a4GmWIWJ21uUSfwfEvySWtC2XADZ4nB+BLYgVIk60CpiwsZ3G93vUEIO3IyNoH/f0wYK8m9TrDHudUZg3qX4waLG5M43q7Wgc/MbQITxOUSQv7c7ugFFDzQGBzZswY6786m86gpEIbb3OhjZnzcvQAaRHhdlQWIMm2nrAgMBAAEwDQYJKoZIhvcNAQEFBQADgYEAtCu4nUhVVxYUntneD9+h8Mg9q6q+auNKyExzyLwaxlAoo7TJHidbtS4J5iNmZgXL0FkbFFBjvSfpJIlJ00zbhNYS5f6GuoEDmFJl0ZxBHjJnyp378OD8uTs7fLvjx79LjSTbNYiytVbZPQUQ5Yaxu2jXnimvw3rrszlaEXAMPLE="
+            },
+            "continuationToken": "A continuation token if continuing job"
+        }
+    }
+  }

examples/event_sources/src/debugging_output.json

@@ -0,0 +1,50 @@
+{
+    "account_id":"111111111111",
+    "data":{
+       "action_configuration":{
+          "configuration":{
+             "decoded_user_parameters":"[Cannot be deserialized]",
+             "function_name":"MyLambdaFunctionForAWSCodePipeline",
+             "raw_event":"[SENSITIVE]",
+             "user_parameters":"some-input-such-as-a-URL"
+          },
+          "raw_event":"[SENSITIVE]"
+       },
+       "artifact_credentials":{
+          "access_key_id":"AKIAIOSFODNN7EXAMPLE",
+          "expiration_time":"None",
+          "raw_event":"[SENSITIVE]",
+          "secret_access_key":"[SENSITIVE]",
+          "session_token":"[SENSITIVE]"
+       },
+       "continuation_token":"A continuation token if continuing job",
+       "encryption_key":"None",
+       "input_artifacts":[
+          {
+             "location":{
+                "get_type":"S3",
+                "raw_event":"[SENSITIVE]",
+                "s3_location":{
+                   "bucket_name":"the name of the bucket configured as the pipeline artifact store in Amazon S3, for example codepipeline-us-east-2-1234567890",
+                   "key":"the name of the application, for example CodePipelineDemoApplication.zip",
+                   "object_key":"the name of the application, for example CodePipelineDemoApplication.zip",
+                   "raw_event":"[SENSITIVE]"
+                }
+             },
+             "name":"ArtifactName",
+             "raw_event":"[SENSITIVE]",
+             "revision":"None"
+          }
+       ],
+       "output_artifacts":[
+
+       ],
+       "raw_event":"[SENSITIVE]"
+    },
+    "decoded_user_parameters":"[Cannot be deserialized]",
+    "get_id":"11111111-abcd-1111-abcd-111111abcdef",
+    "input_bucket_name":"the name of the bucket configured as the pipeline artifact store in Amazon S3, for example codepipeline-us-east-2-1234567890",
+    "input_object_key":"the name of the application, for example CodePipelineDemoApplication.zip",
+    "raw_event":"[SENSITIVE]",
+    "user_parameters":"some-input-such-as-a-URL"
+ }

tests/functional/test_data_classes.py

@@ -126,6 +126,150 @@ class DataClassSample(DictWrapper):
     assert event_source.items() == data.items()
 
 
+def test_dict_wrapper_str_no_property():
+    """
+    Checks that the _properties function returns
+    only the "raw_event", and the resulting string
+    notes it as sensitive.
+    """
+
+    class DataClassSample(DictWrapper):
+        attribute = None
+
+        def function(self) -> None:
+            pass
+
+    event_source = DataClassSample({})
+    assert str(event_source) == "{'raw_event': '[SENSITIVE]'}"
+
+
+def test_dict_wrapper_str_single_property():
+    """
+    Checks that the _properties function returns
+    the defined property "data_property", and
+    resulting string includes the property value.
+    """
+
+    class DataClassSample(DictWrapper):
+        attribute = None
+
+        def function(self) -> None:
+            pass
+
+        @property
+        def data_property(self) -> str:
+            return "value"
+
+    event_source = DataClassSample({})
+    assert str(event_source) == "{'data_property': 'value', 'raw_event': '[SENSITIVE]'}"
+
+
+def test_dict_wrapper_str_property_exception():
+    """
+    Check the recursive _str_helper function handles
+    exceptions that may occur when accessing properties
+    """
+
+    class DataClassSample(DictWrapper):
+        attribute = None
+
+        def function(self) -> None:
+            pass
+
+        @property
+        def data_property(self):
+            raise Exception()
+
+    event_source = DataClassSample({})
+    assert str(event_source) == "{'data_property': '[Cannot be deserialized]', 'raw_event': '[SENSITIVE]'}"
+
+
+def test_dict_wrapper_str_property_list_exception():
+    """
+    Check that _str_helper properly handles exceptions
+    that occur when recursively working through items
+    in a list property.
+    """
+
+    class BrokenDataClass(DictWrapper):
+        @property
+        def broken_data_property(self):
+            raise Exception()
+
+    class DataClassSample(DictWrapper):
+        attribute = None
+
+        def function(self) -> None:
+            pass
+
+        @property
+        def data_property(self) -> list:
+            return ["string", 0, 0.0, BrokenDataClass({})]
+
+    event_source = DataClassSample({})
+    event_str = (
+        "{'data_property': ['string', 0, 0.0, {'broken_data_property': "
+        + "'[Cannot be deserialized]', 'raw_event': '[SENSITIVE]'}], 'raw_event': '[SENSITIVE]'}"
+    )
+    assert str(event_source) == event_str
+
+
+def test_dict_wrapper_str_recursive_property():
+    """
+    Check that the _str_helper function recursively
+    handles Data Classes within Data Classes
+    """
+
+    class DataClassTerminal(DictWrapper):
+        attribute = None
+
+        def function(self) -> None:
+            pass
+
+        @property
+        def terminal_property(self) -> str:
+            return "end-recursion"
+
+    class DataClassRecursive(DictWrapper):
+        attribute = None
+
+        def function(self) -> None:
+            pass
+
+        @property
+        def data_property(self) -> DataClassTerminal:
+            return DataClassTerminal({})
+
+    event_source = DataClassRecursive({})
+    assert (
+        str(event_source)
+        == "{'data_property': {'raw_event': '[SENSITIVE]', 'terminal_property': 'end-recursion'},"
+        + " 'raw_event': '[SENSITIVE]'}"
+    )
+
+
+def test_dict_wrapper_sensitive_properties_property():
+    """
+    Checks that the _str_helper function correctly
+    handles _sensitive_properties
+    """
+
+    class DataClassSample(DictWrapper):
+        attribute = None
+
+        def function(self) -> None:
+            pass
+
+        _sensitive_properties = ["data_property"]
+
+        @property
+        def data_property(self) -> str:
+            return "value"
+
+    event_source = DataClassSample({})
+    assert str(event_source) == "{'data_property': '[SENSITIVE]', 'raw_event': '[SENSITIVE]'}"
+
+
 def test_cloud_watch_dashboard_event():
     event = CloudWatchDashboardCustomWidgetEvent(load_event("cloudWatchDashboardEvent.json"))
     assert event.describe is False