feat(data-classes): add AttributeValueType to DynamoDBStreamEvent

aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py

@@ -1,15 +1,39 @@
 from enum import Enum
-from typing import Dict, Iterator, List, Optional
+from typing import Any, Dict, Iterator, List, Optional, Union
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
 
+class AttributeValueType(Enum):
+    Binary = "B"
+    BinarySet = "BS"
+    Boolean = "BOOL"
+    List = "L"
+    Map = "M"
+    Number = "N"
+    NumberSet = "NS"
+    Null = "NULL"
+    String = "S"
+    StringSet = "SS"
+
+
 class AttributeValue(DictWrapper):
     """Represents the data for an attribute
 
     Documentation: https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_AttributeValue.html
     """
 
+    def __init__(self, data: Dict[str, Any]):
+        """AttributeValue constructor
+
+        Parameters
+        ----------
+        data: Dict[str, Any]
+            Raw lambda event dict
+        """
+        super().__init__(data)
+        self.dynamodb_type = list(data.keys())[0]
+
     @property
     def b_value(self) -> Optional[str]:
         """An attribute of type Base64-encoded binary data object
@@ -106,6 +130,29 @@ def ss_value(self) -> Optional[List[str]]:
         """
         return self.get("SS")
 
+    @property
+    def get_type(self) -> AttributeValueType:
+        """Get the attribute value type based on the contained data"""
+        return AttributeValueType(self.dynamodb_type)
+
+    @property
+    def value(self) -> Union[Optional[bool], Optional[str], Optional[List], Optional[Dict]]:
+        """Get the attribute value"""
+        try:
+            return getattr(self, f"{self.dynamodb_type.lower()}_value")
+        except AttributeError:
+            raise TypeError(f"Dynamodb type {self.dynamodb_type} is not supported")
+
+    @property
+    def l_value(self) -> Optional[List["AttributeValue"]]:
+        """Alias of list_value"""
+        return self.list_value
+
+    @property
+    def m_value(self) -> Optional[Dict[str, "AttributeValue"]]:
+        """Alias of map_value"""
+        return self.map_value
+
 
 def _attribute_value_dict(attr_values: Dict[str, dict], key: str) -> Optional[Dict[str, AttributeValue]]:
     """A dict of type String to AttributeValue object map

tests/functional/test_data_classes.py

@@ -58,6 +58,7 @@
 )
 from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
     AttributeValue,
+    AttributeValueType,
     DynamoDBRecordEventName,
     DynamoDBStreamEvent,
     StreamViewType,
@@ -443,13 +444,43 @@ def test_dynamo_db_stream_trigger_event():
     assert record.user_identity is None
 
 
+def test_dynamo_attribute_value_b_value():
+    example_attribute_value = {"B": "dGhpcyB0ZXh0IGlzIGJhc2U2NC1lbmNvZGVk"}
+
+    attribute_value = AttributeValue(example_attribute_value)
+
+    assert attribute_value.get_type == AttributeValueType.Binary
+    assert attribute_value.b_value == attribute_value.value
+
+
+def test_dynamo_attribute_value_bs_value():
+    example_attribute_value = {"BS": ["U3Vubnk=", "UmFpbnk=", "U25vd3k="]}
+
+    attribute_value = AttributeValue(example_attribute_value)
+
+    assert attribute_value.get_type == AttributeValueType.BinarySet
+    assert attribute_value.bs_value == attribute_value.value
+
+
+def test_dynamo_attribute_value_bool_value():
+    example_attribute_value = {"BOOL": True}
+
+    attribute_value = AttributeValue(example_attribute_value)
+
+    assert attribute_value.get_type == AttributeValueType.Boolean
+    assert attribute_value.bool_value == attribute_value.value
+
+
 def test_dynamo_attribute_value_list_value():
     example_attribute_value = {"L": [{"S": "Cookies"}, {"S": "Coffee"}, {"N": "3.14159"}]}
     attribute_value = AttributeValue(example_attribute_value)
     list_value = attribute_value.list_value
     assert list_value is not None
     item = list_value[0]
     assert item.s_value == "Cookies"
+    assert attribute_value.get_type == AttributeValueType.List
+    assert attribute_value.l_value == attribute_value.list_value
+    assert attribute_value.list_value == attribute_value.value
 
 
 def test_dynamo_attribute_value_map_value():
@@ -461,6 +492,65 @@ def test_dynamo_attribute_value_map_value():
     assert map_value is not None
     item = map_value["Name"]
     assert item.s_value == "Joe"
+    assert attribute_value.get_type == AttributeValueType.Map
+    assert attribute_value.m_value == attribute_value.map_value
+    assert attribute_value.map_value == attribute_value.value
+
+
+def test_dynamo_attribute_value_n_value():
+    example_attribute_value = {"N": "123.45"}
+
+    attribute_value = AttributeValue(example_attribute_value)
+
+    assert attribute_value.get_type == AttributeValueType.Number
+    assert attribute_value.n_value == attribute_value.value
+
+
+def test_dynamo_attribute_value_ns_value():
+    example_attribute_value = {"NS": ["42.2", "-19", "7.5", "3.14"]}
+
+    attribute_value = AttributeValue(example_attribute_value)
+
+    assert attribute_value.get_type == AttributeValueType.NumberSet
+    assert attribute_value.ns_value == attribute_value.value
+
+
+def test_dynamo_attribute_value_null_value():
+    example_attribute_value = {"NULL": True}
+
+    attribute_value = AttributeValue(example_attribute_value)
+
+    assert attribute_value.get_type == AttributeValueType.Null
+    assert attribute_value.null_value == attribute_value.value
+
+
+def test_dynamo_attribute_value_s_value():
+    example_attribute_value = {"S": "Hello"}
+
+    attribute_value = AttributeValue(example_attribute_value)
+
+    assert attribute_value.get_type == AttributeValueType.String
+    assert attribute_value.s_value == attribute_value.value
+
+
+def test_dynamo_attribute_value_ss_value():
+    example_attribute_value = {"SS": ["Giraffe", "Hippo", "Zebra"]}
+
+    attribute_value = AttributeValue(example_attribute_value)
+
+    assert attribute_value.get_type == AttributeValueType.StringSet
+    assert attribute_value.ss_value == attribute_value.value
+
+
+def test_dynamo_attribute_value_type_error():
+    example_attribute_value = {"UNSUPPORTED": "'value' should raise a type error"}
+
+    attribute_value = AttributeValue(example_attribute_value)
+
+    with pytest.raises(TypeError):
+        print(attribute_value.value)
+    with pytest.raises(ValueError):
+        print(attribute_value.get_type)
 
 
 def test_event_bridge_event():