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

aws_lambda_powertools/utilities/data_classes/common.py

@@ -55,14 +55,16 @@ def _str_helper(self) -> Dict[str, Any]:
                     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 as e:
-                    result[property_key] = f"[EXCEPTION {type(e)}]"
+                except Exception:
+                    result[property_key] = "[Cannot be deserialized]"
 
         return result
 

docs/utilities/data_classes.md

@@ -12,7 +12,6 @@ Event Source Data Classes utility provides classes self-describing Lambda event
 * Type hinting and code completion for common event types
 * Helper functions for decoding/deserializing nested fields
 * Docstrings for fields contained in event schemas
-* Implement str() to recursively inspect contents
 
 **Background**
 
@@ -1121,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, like JSON dictionaries, can't be processed and will display as "[Cannot be deserialized]""
+
+=== "debugging.py"
+    ```python hl_lines="5"
+    --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

@@ -136,11 +136,10 @@ def test_dict_wrapper_str_no_property():
     class DataClassSample(DictWrapper):
         attribute = None
 
-        def function(self):
+        def function(self) -> None:
             pass
 
     event_source = DataClassSample({})
-    assert event_source._properties() == ["raw_event"]
     assert str(event_source) == "{'raw_event': '[SENSITIVE]'}"
 
 
@@ -154,16 +153,14 @@ def test_dict_wrapper_str_single_property():
     class DataClassSample(DictWrapper):
         attribute = None
 
-        def function(self):
+        def function(self) -> None:
             pass
 
         @property
-        def data_property(self):
+        def data_property(self) -> str:
             return "value"
 
     event_source = DataClassSample({})
-    assert event_source._properties() == ["data_property", "raw_event"]
-    assert event_source._str_helper() == {"data_property": "value", "raw_event": "[SENSITIVE]"}
     assert str(event_source) == "{'data_property': 'value', 'raw_event': '[SENSITIVE]'}"
 
 
@@ -176,19 +173,15 @@ def test_dict_wrapper_str_property_exception():
     class DataClassSample(DictWrapper):
         attribute = None
 
-        def function(self):
+        def function(self) -> None:
             pass
 
         @property
         def data_property(self):
             raise Exception()
 
     event_source = DataClassSample({})
-    assert event_source._str_helper() == {
-        "data_property": "[EXCEPTION <class 'Exception'>]",
-        "raw_event": "[SENSITIVE]",
-    }
-    assert str(event_source) == "{'data_property': \"[EXCEPTION <class 'Exception'>]\", 'raw_event': '[SENSITIVE]'}"
+    assert str(event_source) == "{'data_property': '[Cannot be deserialized]', 'raw_event': '[SENSITIVE]'}"
 
 
 def test_dict_wrapper_str_property_list_exception():
@@ -206,27 +199,17 @@ def broken_data_property(self):
     class DataClassSample(DictWrapper):
         attribute = None
 
-        def function(self):
+        def function(self) -> None:
             pass
 
         @property
-        def data_property(self):
+        def data_property(self) -> list:
             return ["string", 0, 0.0, BrokenDataClass({})]
 
     event_source = DataClassSample({})
-    assert event_source._properties() == ["data_property", "raw_event"]
-    assert event_source._str_helper() == {
-        "data_property": [
-            "string",
-            0,
-            0.0,
-            {"broken_data_property": "[EXCEPTION <class 'Exception'>]", "raw_event": "[SENSITIVE]"},
-        ],
-        "raw_event": "[SENSITIVE]",
-    }
     event_str = (
         "{'data_property': ['string', 0, 0.0, {'broken_data_property': "
-        + "\"[EXCEPTION <class 'Exception'>]\", 'raw_event': '[SENSITIVE]'}], 'raw_event': '[SENSITIVE]'}"
+        + "'[Cannot be deserialized]', 'raw_event': '[SENSITIVE]'}], 'raw_event': '[SENSITIVE]'}"
     )
     assert str(event_source) == event_str
 
@@ -240,29 +223,24 @@ def test_dict_wrapper_str_recursive_property():
     class DataClassTerminal(DictWrapper):
         attribute = None
 
-        def function(self):
+        def function(self) -> None:
             pass
 
         @property
-        def terminal_property(self):
+        def terminal_property(self) -> str:
             return "end-recursion"
 
     class DataClassRecursive(DictWrapper):
         attribute = None
 
-        def function(self):
+        def function(self) -> None:
             pass
 
         @property
-        def data_property(self):
+        def data_property(self) -> DataClassTerminal:
             return DataClassTerminal({})
 
     event_source = DataClassRecursive({})
-    assert event_source._properties() == ["data_property", "raw_event"]
-    assert event_source._str_helper() == {
-        "data_property": {"raw_event": "[SENSITIVE]", "terminal_property": "end-recursion"},
-        "raw_event": "[SENSITIVE]",
-    }
     assert (
         str(event_source)
         == "{'data_property': {'raw_event': '[SENSITIVE]', 'terminal_property': 'end-recursion'},"
@@ -279,18 +257,16 @@ def test_dict_wrapper_sensitive_properties_property():
     class DataClassSample(DictWrapper):
         attribute = None
 
-        def function(self):
+        def function(self) -> None:
             pass
 
         _sensitive_properties = ["data_property"]
 
         @property
-        def data_property(self):
+        def data_property(self) -> str:
             return "value"
 
     event_source = DataClassSample({})
-    assert event_source._properties() == ["data_property", "raw_event"]
-    assert event_source._str_helper() == {"data_property": "[SENSITIVE]", "raw_event": "[SENSITIVE]"}
     assert str(event_source) == "{'data_property': '[SENSITIVE]', 'raw_event': '[SENSITIVE]'}"