Merge branch 'develop' of https://github.com/awslabs/aws-lambda-powertools-python into develop

* 'develop' of https://github.com/awslabs/aws-lambda-powertools-python:
  chore(deps): bump boto3 from 1.17.88 to 1.17.89 (#466)
  feat(data-classes): add AttributeValueType to DynamoDBStreamEvent (#462)
  chore(deps): bump boto3 from 1.17.87 to 1.17.88 (#463)
  chore(deps-dev): bump mkdocs-material from 7.1.6 to 7.1.7 (#464)
  feat(data-classes): decorator to instantiate data_classes and docs updates (#442)
  chore(deps): bump boto3 from 1.17.86 to 1.17.87 (#459)

Author: heitorlessa

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -10,6 +10,7 @@
 from .connect_contact_flow_event import ConnectContactFlowEvent
 from .dynamo_db_stream_event import DynamoDBStreamEvent
 from .event_bridge_event import EventBridgeEvent
+from .event_source import event_source
 from .kinesis_stream_event import KinesisStreamEvent
 from .s3_event import S3Event
 from .ses_event import SESEvent
@@ -31,4 +32,5 @@
     "SESEvent",
     "SNSEvent",
     "SQSEvent",
+    "event_source",
 ]

aws_lambda_powertools/utilities/data_classes/alb_event.py

@@ -6,6 +6,7 @@
 class ALBEventRequestContext(DictWrapper):
     @property
     def elb_target_group_arn(self) -> str:
+        """Target group arn for your Lambda function"""
         return self["requestContext"]["elb"]["targetGroupArn"]
 
 
@@ -15,6 +16,7 @@ class ALBEvent(BaseProxyEvent):
     Documentation:
     --------------
     - https://docs.aws.amazon.com/lambda/latest/dg/services-alb.html
+    - https://docs.aws.amazon.com/elasticloadbalancing/latest/application/lambda-functions.html
     """
 
     @property

aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py

@@ -1,15 +1,42 @@
 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
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_AttributeValue.html
+    - https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.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 +133,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 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
+
+    @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 _attribute_value_dict(attr_values: Dict[str, dict], key: str) -> Optional[Dict[str, AttributeValue]]:
     """A dict of type String to AttributeValue object map
@@ -224,6 +274,29 @@ class DynamoDBStreamEvent(DictWrapper):
     Documentation:
     -------------
     - https://docs.aws.amazon.com/lambda/latest/dg/with-ddb.html
+
+    Example
+    -------
+    **Process dynamodb stream events and use get_type and get_value for handling conversions**
+
+        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)
     """
 
     @property

aws_lambda_powertools/utilities/data_classes/event_source.py

@@ -0,0 +1,39 @@
+from typing import Any, Callable, Dict, Type
+
+from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+
+@lambda_handler_decorator
+def event_source(
+    handler: Callable[[Any, LambdaContext], Any],
+    event: Dict[str, Any],
+    context: LambdaContext,
+    data_class: Type[DictWrapper],
+):
+    """Middleware to create an instance of the passed in event source data class
+
+    Parameters
+    ----------
+    handler: Callable
+        Lambda's handler
+    event: Dict
+        Lambda's Event
+    context: Dict
+        Lambda's Context
+    data_class: Type[DictWrapper]
+        Data class type to instantiate
+
+    Example
+    --------
+
+    **Sample usage**
+
+        from aws_lambda_powertools.utilities.data_classes import S3Event, event_source
+
+        @event_source(data_class=S3Event)
+        def handler(event: S3Event, context):
+             return {"key": event.object_key}
+    """
+    return handler(data_class(event), context)

aws_lambda_powertools/utilities/idempotency/persistence/base.py

@@ -224,6 +224,7 @@ def _generate_hash(self, data: Any) -> str:
             Hashed representation of the provided data
 
         """
+        data = getattr(data, "raw_event", data)  # could be a data class depending on decorator order
         hashed_data = self.hash_function(json.dumps(data, cls=Encoder).encode())
         return hashed_data.hexdigest()