fix(data_sources): ensure correct types on SQSMessageAttributes

aws_lambda_powertools/utilities/data_classes/sqs_event.py

@@ -1,5 +1,5 @@
 from functools import cached_property
-from typing import Any, Dict, Iterator, Optional, Type, TypeVar
+from typing import Any, Dict, ItemsView, Iterator, Optional, Type, TypeVar
 
 from aws_lambda_powertools.utilities.data_classes import S3Event
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
@@ -82,6 +82,9 @@ def __getitem__(self, key: str) -> Optional[SQSMessageAttribute]:  # type: ignor
         item = super().get(key)
         return None if item is None else SQSMessageAttribute(item)  # type: ignore
 
+    def items(self) -> ItemsView[str, SQSMessageAttribute]:  # type: ignore
+        return {k: SQSMessageAttribute(v) for k, v in super().items()}.items()  # type: ignore
+
 
 class SQSRecord(DictWrapper):
     """An Amazon SQS message"""

tests/unit/data_classes/test_sqs_event.py

@@ -2,6 +2,7 @@
 
 from aws_lambda_powertools.utilities.data_classes import S3Event, SQSEvent
 from aws_lambda_powertools.utilities.data_classes.sns_event import SNSMessage
+from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSMessageAttributes
 from tests.functional.utils import load_event
 
 
@@ -132,3 +133,10 @@ def test_decode_nested_sns_event():
     raw_message = json.loads(raw_body["Message"])
     assert message["message"] == raw_message["message"]
     assert message["username"] == raw_message["username"]
+
+
+def test_sqs_event_typing():
+    attributes = SQSMessageAttributes({"key": {"stringValue": "value", "dataType": "String"}})
+
+    # This assertion compares the return from .items() to the return of __getitem__
+    assert list(attributes.items())[0][1] == attributes["key"]