diff --git a/aws_lambda_powertools/utilities/batch/base.py b/aws_lambda_powertools/utilities/batch/base.py
index e4a869a1e54..4f9c4ca8780 100644
--- a/aws_lambda_powertools/utilities/batch/base.py
+++ b/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
diff --git a/aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py b/aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py
index eb674c86b60..e62e307d67a 100644
--- a/aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py
+++ b/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
diff --git a/docs/upgrade.md b/docs/upgrade.md
index f6b3c7e9d00..fcce2f1958d 100644
--- a/docs/upgrade.md
+++ b/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
+
+```
diff --git a/docs/utilities/batch.md b/docs/utilities/batch.md
index 1bbba86c395..7fcf1ff46d8 100644
--- a/docs/utilities/batch.md
+++ b/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
diff --git a/docs/utilities/data_classes.md b/docs/utilities/data_classes.md
index 67d821fe04f..4ab41d30d7f 100644
--- a/docs/utilities/data_classes.md
+++ b/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
diff --git a/tests/functional/test_data_classes.py b/tests/functional/test_data_classes.py
index 1f8c0cef955..4fe0eb40331 100644
--- a/tests/functional/test_data_classes.py
+++ b/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()
diff --git a/tests/functional/test_utilities_batch.py b/tests/functional/test_utilities_batch.py
index 4f46b428121..1d50de9e85e 100644
--- a/tests/functional/test_utilities_batch.py
+++ b/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