feat(data-classes): replace AttributeValue in DynamoDBStreamEvent with deserialized Python values (#1619)

Co-authored-by: heitorlessa <[email protected]>

Author: shanab

aws_lambda_powertools/utilities/batch/base.py

@@ -323,10 +323,10 @@ def lambda_handler(event, context: LambdaContext):
     @tracer.capture_method
     def record_handler(record: DynamoDBRecord):
         logger.info(record.dynamodb.new_image)
-        payload: dict = json.loads(record.dynamodb.new_image.get("item").s_value)
+        payload: dict = json.loads(record.dynamodb.new_image.get("item"))
         # alternatively:
-        # changes: Dict[str, dynamo_db_stream_event.AttributeValue] = record.dynamodb.new_image  # noqa: E800
-        # payload = change.get("Message").raw_event -> {"S": "<payload>"}
+        # changes: Dict[str, Any] = record.dynamodb.new_image  # noqa: E800
+        # payload = change.get("Message") -> "<payload>"
         ...
 
     @logger.inject_lambda_context

aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py

@@ -1,169 +1,100 @@
+from decimal import Clamped, Context, Decimal, Inexact, Overflow, Rounded, Underflow
 from enum import Enum
-from typing import Any, Dict, Iterator, List, Optional, Union
+from typing import Any, Callable, Dict, Iterator, Optional, Sequence, Set
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
+# NOTE: DynamoDB supports up to 38 digits precision
+# Therefore, this ensures our Decimal follows what's stored in the table
+DYNAMODB_CONTEXT = Context(
+    Emin=-128,
+    Emax=126,
+    prec=38,
+    traps=[Clamped, Overflow, Inexact, Rounded, Underflow],
+)
 
-class AttributeValueType(Enum):
-    Binary = "B"
-    BinarySet = "BS"
-    Boolean = "BOOL"
-    List = "L"
-    Map = "M"
-    Number = "N"
-    NumberSet = "NS"
-    Null = "NULL"
-    String = "S"
-    StringSet = "SS"
 
+class TypeDeserializer:
+    """
+    Deserializes DynamoDB types to Python types.
 
-class AttributeValue(DictWrapper):
-    """Represents the data for an attribute
+    It's based on boto3's [DynamoDB TypeDeserializer](https://boto3.amazonaws.com/v1/documentation/api/latest/_modules/boto3/dynamodb/types.html). # noqa: E501
 
-    Documentation:
-    --------------
-    - https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_AttributeValue.html
-    - https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html
+    The only notable difference is that for Binary (`B`, `BS`) values we return Python Bytes directly,
+    since we don't support Python 2.
     """
 
-    def __init__(self, data: Dict[str, Any]):
-        """AttributeValue constructor
+    def deserialize(self, value: Dict) -> Any:
+        """Deserialize DynamoDB data types into Python types.
 
         Parameters
         ----------
-        data: Dict[str, Any]
-            Raw lambda event dict
-        """
-        super().__init__(data)
-        self.dynamodb_type = list(data.keys())[0]
+        value: Any
+            DynamoDB value to be deserialized to a python type
 
-    @property
-    def b_value(self) -> Optional[str]:
-        """An attribute of type Base64-encoded binary data object
 
-        Example:
-            >>> {"B": "dGhpcyB0ZXh0IGlzIGJhc2U2NC1lbmNvZGVk"}
-        """
-        return self.get("B")
+            Here are the various conversions:
 
-    @property
-    def bs_value(self) -> Optional[List[str]]:
-        """An attribute of type Array of Base64-encoded binary data objects
-
-        Example:
-            >>> {"BS": ["U3Vubnk=", "UmFpbnk=", "U25vd3k="]}
-        """
-        return self.get("BS")
-
-    @property
-    def bool_value(self) -> Optional[bool]:
-        """An attribute of type Boolean
-
-        Example:
-            >>> {"BOOL": True}
-        """
-        item = self.get("BOOL")
-        return None if item is None else bool(item)
+            DynamoDB                                Python
+            --------                                ------
+            {'NULL': True}                          None
+            {'BOOL': True/False}                    True/False
+            {'N': str(value)}                       str(value)
+            {'S': string}                           string
+            {'B': bytes}                            bytes
+            {'NS': [str(value)]}                    set([str(value)])
+            {'SS': [string]}                        set([string])
+            {'BS': [bytes]}                         set([bytes])
+            {'L': list}                             list
+            {'M': dict}                             dict
 
-    @property
-    def list_value(self) -> Optional[List["AttributeValue"]]:
-        """An attribute of type Array of AttributeValue objects
-
-        Example:
-            >>> {"L": [ {"S": "Cookies"} , {"S": "Coffee"}, {"N": "3.14159"}]}
-        """
-        item = self.get("L")
-        return None if item is None else [AttributeValue(v) for v in item]
-
-    @property
-    def map_value(self) -> Optional[Dict[str, "AttributeValue"]]:
-        """An attribute of type String to AttributeValue object map
-
-        Example:
-            >>> {"M": {"Name": {"S": "Joe"}, "Age": {"N": "35"}}}
-        """
-        return _attribute_value_dict(self._data, "M")
-
-    @property
-    def n_value(self) -> Optional[str]:
-        """An attribute of type Number
-
-        Numbers are sent across the network to DynamoDB as strings, to maximize compatibility across languages
-        and libraries. However, DynamoDB treats them as number type attributes for mathematical operations.
+        Parameters
+        ----------
+        value: Any
+            DynamoDB value to be deserialized to a python type
 
-        Example:
-            >>> {"N": "123.45"}
+        Returns
+        --------
+        any
+            Python native type converted from DynamoDB type
         """
-        return self.get("N")
-
-    @property
-    def ns_value(self) -> Optional[List[str]]:
-        """An attribute of type Number Set
 
-        Example:
-            >>> {"NS": ["42.2", "-19", "7.5", "3.14"]}
-        """
-        return self.get("NS")
+        dynamodb_type = list(value.keys())[0]
+        deserializer: Optional[Callable] = getattr(self, f"_deserialize_{dynamodb_type}".lower(), None)
+        if deserializer is None:
+            raise TypeError(f"Dynamodb type {dynamodb_type} is not supported")
 
-    @property
-    def null_value(self) -> None:
-        """An attribute of type Null.
+        return deserializer(value[dynamodb_type])
 
-        Example:
-            >>> {"NULL": True}
-        """
+    def _deserialize_null(self, value: bool) -> None:
         return None
 
-    @property
-    def s_value(self) -> Optional[str]:
-        """An attribute of type String
+    def _deserialize_bool(self, value: bool) -> bool:
+        return value
 
-        Example:
-            >>> {"S": "Hello"}
-        """
-        return self.get("S")
+    def _deserialize_n(self, value: str) -> Decimal:
+        return DYNAMODB_CONTEXT.create_decimal(value)
 
-    @property
-    def ss_value(self) -> Optional[List[str]]:
-        """An attribute of type Array of strings
+    def _deserialize_s(self, value: str) -> str:
+        return value
 
-        Example:
-            >>> {"SS": ["Giraffe", "Hippo" ,"Zebra"]}
-        """
-        return self.get("SS")
+    def _deserialize_b(self, value: bytes) -> bytes:
+        return value
 
-    @property
-    def get_type(self) -> AttributeValueType:
-        """Get the attribute value type based on the contained data"""
-        return AttributeValueType(self.dynamodb_type)
+    def _deserialize_ns(self, value: Sequence[str]) -> Set[Decimal]:
+        return set(map(self._deserialize_n, value))
 
-    @property
-    def l_value(self) -> Optional[List["AttributeValue"]]:
-        """Alias of list_value"""
-        return self.list_value
+    def _deserialize_ss(self, value: Sequence[str]) -> Set[str]:
+        return set(map(self._deserialize_s, value))
 
-    @property
-    def m_value(self) -> Optional[Dict[str, "AttributeValue"]]:
-        """Alias of map_value"""
-        return self.map_value
+    def _deserialize_bs(self, value: Sequence[bytes]) -> Set[bytes]:
+        return set(map(self._deserialize_b, value))
 
-    @property
-    def get_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")
+    def _deserialize_l(self, value: Sequence[Dict]) -> Sequence[Any]:
+        return [self.deserialize(v) for v in 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
-
-    Example:
-        >>> {"NewImage": {"Id": {"S": "xxx-xxx"}, "Value": {"N": "35"}}}
-    """
-    attr_values_dict = attr_values.get(key)
-    return None if attr_values_dict is None else {k: AttributeValue(v) for k, v in attr_values_dict.items()}
+    def _deserialize_m(self, value: Dict) -> Dict:
+        return {k: self.deserialize(v) for k, v in value.items()}
 
 
 class StreamViewType(Enum):
@@ -176,28 +107,57 @@ class StreamViewType(Enum):
 
 
 class StreamRecord(DictWrapper):
+    _deserializer = TypeDeserializer()
+
+    def __init__(self, data: Dict[str, Any]):
+        """StreamRecord constructor
+        Parameters
+        ----------
+        data: Dict[str, Any]
+            Represents the dynamodb dict inside DynamoDBStreamEvent's records
+        """
+        super().__init__(data)
+        self._deserializer = TypeDeserializer()
+
+    def _deserialize_dynamodb_dict(self, key: str) -> Optional[Dict[str, Any]]:
+        """Deserialize DynamoDB records available in `Keys`, `NewImage`, and `OldImage`
+
+        Parameters
+        ----------
+        key : str
+            DynamoDB key (e.g., Keys, NewImage, or OldImage)
+
+        Returns
+        -------
+        Optional[Dict[str, Any]]
+            Deserialized records in Python native types
+        """
+        dynamodb_dict = self._data.get(key)
+        if dynamodb_dict is None:
+            return None
+
+        return {k: self._deserializer.deserialize(v) for k, v in dynamodb_dict.items()}
+
     @property
     def approximate_creation_date_time(self) -> Optional[int]:
         """The approximate date and time when the stream record was created, in UNIX epoch time format."""
         item = self.get("ApproximateCreationDateTime")
         return None if item is None else int(item)
 
-    # NOTE: This override breaks the Mapping protocol of DictWrapper, it's left here for backwards compatibility with
-    # a 'type: ignore' comment. See #1516 for discussion
     @property
-    def keys(self) -> Optional[Dict[str, AttributeValue]]:  # type: ignore[override]
+    def keys(self) -> Optional[Dict[str, Any]]:  # type: ignore[override]
         """The primary key attribute(s) for the DynamoDB item that was modified."""
-        return _attribute_value_dict(self._data, "Keys")
+        return self._deserialize_dynamodb_dict("Keys")
 
     @property
-    def new_image(self) -> Optional[Dict[str, AttributeValue]]:
+    def new_image(self) -> Optional[Dict[str, Any]]:
         """The item in the DynamoDB table as it appeared after it was modified."""
-        return _attribute_value_dict(self._data, "NewImage")
+        return self._deserialize_dynamodb_dict("NewImage")
 
     @property
-    def old_image(self) -> Optional[Dict[str, AttributeValue]]:
+    def old_image(self) -> Optional[Dict[str, Any]]:
         """The item in the DynamoDB table as it appeared before it was modified."""
-        return _attribute_value_dict(self._data, "OldImage")
+        return self._deserialize_dynamodb_dict("OldImage")
 
     @property
     def sequence_number(self) -> Optional[str]:
@@ -233,7 +193,7 @@ def aws_region(self) -> Optional[str]:
 
     @property
     def dynamodb(self) -> Optional[StreamRecord]:
-        """The main body of the stream record, containing all the DynamoDB-specific fields."""
+        """The main body of the stream record, containing all the DynamoDB-specific dicts."""
         stream_record = self.get("dynamodb")
         return None if stream_record is None else StreamRecord(stream_record)
 
@@ -278,26 +238,18 @@ class DynamoDBStreamEvent(DictWrapper):
 
     Example
     -------
-    **Process dynamodb stream events and use get_type and get_value for handling conversions**
+    **Process dynamodb stream events. DynamoDB types are automatically converted to their equivalent Python values.**
 
         from aws_lambda_powertools.utilities.data_classes import event_source, DynamoDBStreamEvent
-        from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
-            AttributeValueType,
-            AttributeValue,
-        )
         from aws_lambda_powertools.utilities.typing import LambdaContext
 
 
         @event_source(data_class=DynamoDBStreamEvent)
         def lambda_handler(event: DynamoDBStreamEvent, context: LambdaContext):
             for record in event.records:
-                key: AttributeValue = record.dynamodb.keys["id"]
-                if key == AttributeValueType.Number:
-                    assert key.get_value == key.n_value
-                    print(key.get_value)
-                elif key == AttributeValueType.Map:
-                    assert key.get_value == key.map_value
-                    print(key.get_value)
+                # {"N": "123.45"} => Decimal("123.45")
+                key: str = record.dynamodb.keys["id"]
+                print(key)
     """
 
     @property

docs/upgrade.md

@@ -14,6 +14,7 @@ Changes at a glance:
 * The **legacy SQS batch processor** was removed.
 * The **Idempotency key** format changed slightly, invalidating all the existing cached results.
 * The **Feature Flags and AppConfig Parameter utility** API calls have changed and you must update your IAM permissions.
+* The **`DynamoDBStreamEvent`** replaced `AttributeValue` with native Python types.
 
 ???+ important
     Powertools for Python v2 drops suport for Python 3.6, following the Python 3.6 End-Of-Life (EOL) reached on December 23, 2021.
@@ -161,3 +162,47 @@ Using qualified names prevents distinct functions with the same name to contend
 AWS AppConfig deprecated the current API (GetConfiguration) - [more details here](https://github.com/awslabs/aws-lambda-powertools-python/issues/1506#issuecomment-1266645884).
 
 You must update your IAM permissions to allow `appconfig:GetLatestConfiguration` and `appconfig:StartConfigurationSession`. There are no code changes required.
+
+## DynamoDBStreamEvent in Event Source Data Classes
+
+???+ info
+    This also applies if you're using [**`BatchProcessor`**](https://awslabs.github.io/aws-lambda-powertools-python/latest/utilities/batch/#processing-messages-from-dynamodb){target="_blank"} to handle DynamoDB Stream events.
+
+You will now receive native Python types when accessing DynamoDB records via `keys`, `new_image`, and `old_image` attributes in `DynamoDBStreamEvent`.
+
+Previously, you'd receive a `AttributeValue` instance and need to deserialize each item to the type you'd want for convenience, or to the type DynamoDB stored via `get_value` method.
+
+With this change, you can access data deserialized as stored in DynamoDB, and no longer need to recursively deserialize nested objects (Maps) if you had them.
+
+???+ note
+    For a lossless conversion of DynamoDB `Number` type, we follow AWS Python SDK (boto3) approach and convert to `Decimal`.
+
+```python hl_lines="15-20 24-25"
+from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
+    DynamoDBStreamEvent,
+    DynamoDBRecordEventName
+)
+
+def send_to_sqs(data: Dict):
+    body = json.dumps(data)
+    ...
+
+@event_source(data_class=DynamoDBStreamEvent)
+def lambda_handler(event: DynamoDBStreamEvent, context):
+    for record in event.records:
+
+        # BEFORE
+        new_image: Dict[str, AttributeValue] = record.dynamodb.new_image
+        event_type: AttributeValue = new_image["eventType"].get_value
+        if event_type == "PENDING":
+            # deserialize attribute value into Python native type
+            # NOTE: nested objects would need additional logic
+            data = {k: v.get_value for k, v in image.items()}
+            send_to_sqs(data)
+
+        # AFTER
+        new_image: Dict[str, Any] = record.dynamodb.new_image
+        if new_image.get("eventType") == "PENDING":
+            send_to_sqs(new_image)  # Here new_image is just a Python Dict type
+
+```

docs/utilities/batch.md

@@ -506,9 +506,9 @@ Processing batches from Kinesis works in four stages:
     @tracer.capture_method
     def record_handler(record: DynamoDBRecord):
         logger.info(record.dynamodb.new_image)
-        payload: dict = json.loads(record.dynamodb.new_image.get("Message").get_value)
+        payload: dict = json.loads(record.dynamodb.new_image.get("Message"))
         # alternatively:
-        # changes: Dict[str, dynamo_db_stream_event.AttributeValue] = record.dynamodb.new_image
+        # changes: Dict[str, Any] = record.dynamodb.new_image
         # payload = change.get("Message").raw_event -> {"S": "<payload>"}
         ...
 
@@ -538,10 +538,10 @@ Processing batches from Kinesis works in four stages:
     @tracer.capture_method
     def record_handler(record: DynamoDBRecord):
         logger.info(record.dynamodb.new_image)
-        payload: dict = json.loads(record.dynamodb.new_image.get("item").s_value)
+        payload: dict = json.loads(record.dynamodb.new_image.get("item"))
         # alternatively:
-        # changes: Dict[str, dynamo_db_stream_event.AttributeValue] = record.dynamodb.new_image
-        # payload = change.get("Message").raw_event -> {"S": "<payload>"}
+        # changes: Dict[str, Any] = record.dynamodb.new_image
+        # payload = change.get("Message") -> "<payload>"
         ...
 
     @logger.inject_lambda_context

docs/utilities/data_classes.md

@@ -797,9 +797,9 @@ This example is based on the AWS Cognito docs for [Verify Auth Challenge Respons
 
 ### DynamoDB Streams
 
-The DynamoDB data class utility provides the base class for `DynamoDBStreamEvent`, a typed class for
-attributes values (`AttributeValue`), as well as enums for stream view type (`StreamViewType`) and event type
+The DynamoDB data class utility provides the base class for `DynamoDBStreamEvent`, as well as enums for stream view type (`StreamViewType`) and event type.
 (`DynamoDBRecordEventName`).
+The class automatically deserializes DynamoDB types into their equivalent Python types.
 
 === "app.py"
 
@@ -823,21 +823,15 @@ attributes values (`AttributeValue`), as well as enums for stream view type (`St
 
     ```python
     from aws_lambda_powertools.utilities.data_classes import event_source, DynamoDBStreamEvent
-    from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import AttributeValueType, AttributeValue
     from aws_lambda_powertools.utilities.typing import LambdaContext
 
 
     @event_source(data_class=DynamoDBStreamEvent)
     def lambda_handler(event: DynamoDBStreamEvent, context: LambdaContext):
         for record in event.records:
-            key: AttributeValue = record.dynamodb.keys["id"]
-            if key == AttributeValueType.Number:
-                # {"N": "123.45"} => "123.45"
-                assert key.get_value == key.n_value
-                print(key.get_value)
-            elif key == AttributeValueType.Map:
-                assert key.get_value == key.map_value
-                print(key.get_value)
+            # {"N": "123.45"} => Decimal("123.45")
+            key: str = record.dynamodb.keys["id"]
+            print(key)
     ```
 
 ### EventBridge

tests/functional/test_data_classes.py

@@ -2,6 +2,7 @@
 import datetime
 import json
 import zipfile
+from decimal import Clamped, Context, Inexact, Overflow, Rounded, Underflow
 from secrets import compare_digest
 from urllib.parse import quote_plus
 
@@ -75,8 +76,6 @@
     ConnectContactFlowInitiationMethod,
 )
 from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
-    AttributeValue,
-    AttributeValueType,
     DynamoDBRecordEventName,
     DynamoDBStreamEvent,
     StreamRecord,
@@ -490,7 +489,13 @@ def test_connect_contact_flow_event_all():
     assert event.parameters == {"ParameterOne": "One", "ParameterTwo": "Two"}
 
 
-def test_dynamo_db_stream_trigger_event():
+def test_dynamodb_stream_trigger_event():
+    decimal_context = Context(
+        Emin=-128,
+        Emax=126,
+        prec=38,
+        traps=[Clamped, Overflow, Inexact, Rounded, Underflow],
+    )
     event = DynamoDBStreamEvent(load_event("dynamoStreamEvent.json"))
 
     records = list(event.records)
@@ -502,20 +507,8 @@ def test_dynamo_db_stream_trigger_event():
     assert dynamodb.approximate_creation_date_time is None
     keys = dynamodb.keys
     assert keys is not None
-    id_key = keys["Id"]
-    assert id_key.b_value is None
-    assert id_key.bs_value is None
-    assert id_key.bool_value is None
-    assert id_key.list_value is None
-    assert id_key.map_value is None
-    assert id_key.n_value == "101"
-    assert id_key.ns_value is None
-    assert id_key.null_value is None
-    assert id_key.s_value is None
-    assert id_key.ss_value is None
-    message_key = dynamodb.new_image["Message"]
-    assert message_key is not None
-    assert message_key.s_value == "New item!"
+    assert keys["Id"] == decimal_context.create_decimal(101)
+    assert dynamodb.new_image["Message"] == "New item!"
     assert dynamodb.old_image is None
     assert dynamodb.sequence_number == "111"
     assert dynamodb.size_bytes == 26
@@ -528,129 +521,61 @@ 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.get_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.get_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.get_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.get_value
-
-
-def test_dynamo_attribute_value_map_value():
-    example_attribute_value = {"M": {"Name": {"S": "Joe"}, "Age": {"N": "35"}}}
-
-    attribute_value = AttributeValue(example_attribute_value)
-
-    map_value = 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.get_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.get_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.get_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 is None
-    assert attribute_value.null_value == attribute_value.get_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.get_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.get_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.get_value)
-    with pytest.raises(ValueError):
-        print(attribute_value.get_type)
-
-
-def test_stream_record_keys_with_valid_keys():
-    attribute_value = {"Foo": "Bar"}
-    record = StreamRecord({"Keys": {"Key1": attribute_value}})
-    assert record.keys == {"Key1": AttributeValue(attribute_value)}
+def test_dynamodb_stream_record_deserialization():
+    byte_list = [s.encode("utf-8") for s in ["item1", "item2"]]
+    decimal_context = Context(
+        Emin=-128,
+        Emax=126,
+        prec=38,
+        traps=[Clamped, Overflow, Inexact, Rounded, Underflow],
+    )
+    data = {
+        "Keys": {"key1": {"attr1": "value1"}},
+        "NewImage": {
+            "Name": {"S": "Joe"},
+            "Age": {"N": "35"},
+            "TypesMap": {
+                "M": {
+                    "string": {"S": "value"},
+                    "number": {"N": "100"},
+                    "bool": {"BOOL": True},
+                    "dict": {"M": {"key": {"S": "value"}}},
+                    "stringSet": {"SS": ["item1", "item2"]},
+                    "numberSet": {"NS": ["100", "200", "300"]},
+                    "binary": {"B": b"\x00"},
+                    "byteSet": {"BS": byte_list},
+                    "list": {"L": [{"S": "item1"}, {"N": "3.14159"}, {"BOOL": False}]},
+                    "null": {"NULL": True},
+                },
+            },
+        },
+    }
+    record = StreamRecord(data)
+    assert record.new_image == {
+        "Name": "Joe",
+        "Age": decimal_context.create_decimal("35"),
+        "TypesMap": {
+            "string": "value",
+            "number": decimal_context.create_decimal("100"),
+            "bool": True,
+            "dict": {"key": "value"},
+            "stringSet": {"item1", "item2"},
+            "numberSet": {decimal_context.create_decimal(n) for n in ["100", "200", "300"]},
+            "binary": b"\x00",
+            "byteSet": set(byte_list),
+            "list": ["item1", decimal_context.create_decimal("3.14159"), False],
+            "null": None,
+        },
+    }
 
 
-def test_stream_record_keys_with_no_keys():
+def test_dynamodb_stream_record_keys_with_no_keys():
     record = StreamRecord({})
     assert record.keys is None
 
 
-def test_stream_record_keys_overrides_dict_wrapper_keys():
-    data = {"Keys": {"key1": {"attr1": "value1"}}}
+def test_dynamodb_stream_record_keys_overrides_dict_wrapper_keys():
+    data = {"Keys": {"key1": {"N": "101"}}}
     record = StreamRecord(data)
     assert record.keys != data.keys()
 

tests/functional/test_utilities_batch.py

@@ -129,7 +129,7 @@ def handler(record: KinesisStreamRecord):
 @pytest.fixture(scope="module")
 def dynamodb_record_handler() -> Callable:
     def handler(record: DynamoDBRecord):
-        body = record.dynamodb.new_image.get("Message").get_value
+        body = record.dynamodb.new_image.get("Message")
         if "fail" in body:
             raise Exception("Failed to process record.")
         return body