feat: Advanced parser utility (pydantic)

Author: Ran Isenberg

aws_lambda_powertools/utilities/advanced_parser/__init__.py

@@ -1,14 +1,6 @@
-"""Validation utility
+"""Advanced parser utility
 """
-from .envelopes import DynamoDBEnvelope, EventBridgeEnvelope, SnsEnvelope, SqsEnvelope, UserEnvelope
-from .validator import validate, validator
+from .envelopes import Envelope, InvalidEnvelopeError, parse_envelope
+from .parser import parser
 
-__all__ = [
-    "UserEnvelope",
-    "DynamoDBEnvelope",
-    "EventBridgeEnvelope",
-    "SnsEnvelope",
-    "SqsEnvelope",
-    "validate",
-    "validator",
-]
+__all__ = ["InvalidEnvelopeError", "Envelope", "parse_envelope", "parser"]

aws_lambda_powertools/utilities/advanced_parser/envelopes/__init__.py

@@ -1,7 +1,3 @@
-from .base import UserEnvelope
-from .dynamodb import DynamoDBEnvelope
-from .event_bridge import EventBridgeEnvelope
-from .sns import SnsEnvelope
-from .sqs import SqsEnvelope
+from .envelopes import Envelope, InvalidEnvelopeError, parse_envelope
 
-__all__ = ["UserEnvelope", "DynamoDBEnvelope", "EventBridgeEnvelope", "SqsEnvelope", "SnsEnvelope"]
+__all__ = ["InvalidEnvelopeError", "Envelope", "parse_envelope"]

aws_lambda_powertools/utilities/advanced_parser/envelopes/base.py

@@ -8,35 +8,26 @@
 
 
 class BaseEnvelope(ABC):
-    def _parse_user_dict_schema(self, user_event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+    def _parse_user_dict_schema(self, user_event: Dict[str, Any], schema: BaseModel) -> Any:
         logger.debug("parsing user dictionary schema")
         try:
-            return inbound_schema_model(**user_event)
+            return schema(**user_event)
         except (ValidationError, TypeError):
             logger.exception("Validation exception while extracting user custom schema")
             raise
 
-    def _parse_user_json_string_schema(self, user_event: str, inbound_schema_model: BaseModel) -> Any:
+    def _parse_user_json_string_schema(self, user_event: str, schema: BaseModel) -> Any:
         logger.debug("parsing user dictionary schema")
-        if inbound_schema_model == str:
+        if schema == str:
             logger.debug("input is string, returning")
             return user_event
         logger.debug("trying to parse as json encoded string")
         try:
-            return inbound_schema_model.parse_raw(user_event)
+            return schema.parse_raw(user_event)
         except (ValidationError, TypeError):
             logger.exception("Validation exception while extracting user custom schema")
             raise
 
     @abstractmethod
-    def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+    def parse(self, event: Dict[str, Any], schema: BaseModel):
         return NotImplemented
-
-
-class UserEnvelope(BaseEnvelope):
-    def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
-        try:
-            return inbound_schema_model(**event)
-        except (ValidationError, TypeError):
-            logger.exception("Validation exception received from input user custom envelopes event")
-            raise

aws_lambda_powertools/utilities/advanced_parser/envelopes/dynamodb.py

@@ -10,7 +10,7 @@
 
 
 class DynamoDBEnvelope(BaseEnvelope):
-    def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+    def parse(self, event: Dict[str, Any], schema: BaseModel) -> Any:
         try:
             parsed_envelope = DynamoDBSchema(**event)
         except (ValidationError, TypeError):
@@ -19,14 +19,10 @@ def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
         output = []
         for record in parsed_envelope.Records:
             parsed_new_image = (
-                {}
-                if not record.dynamodb.NewImage
-                else self._parse_user_dict_schema(record.dynamodb.NewImage, inbound_schema_model)
+                None if not record.dynamodb.NewImage else self._parse_user_dict_schema(record.dynamodb.NewImage, schema)
             )  # noqa: E501
             parsed_old_image = (
-                {}
-                if not record.dynamodb.OldImage
-                else self._parse_user_dict_schema(record.dynamodb.OldImage, inbound_schema_model)
+                None if not record.dynamodb.OldImage else self._parse_user_dict_schema(record.dynamodb.OldImage, schema)
             )  # noqa: E501
-            output.append({"new": parsed_new_image, "old": parsed_old_image})
+            output.append({"NewImage": parsed_new_image, "OldImage": parsed_old_image})
         return output

aws_lambda_powertools/utilities/advanced_parser/envelopes/envelopes.py

@@ -0,0 +1,42 @@
+import logging
+from enum import Enum
+from typing import Any, Dict
+
+from pydantic import BaseModel
+
+from aws_lambda_powertools.utilities.advanced_parser.envelopes.base import BaseEnvelope
+from aws_lambda_powertools.utilities.advanced_parser.envelopes.dynamodb import DynamoDBEnvelope
+from aws_lambda_powertools.utilities.advanced_parser.envelopes.event_bridge import EventBridgeEnvelope
+from aws_lambda_powertools.utilities.advanced_parser.envelopes.sqs import SqsEnvelope
+
+logger = logging.getLogger(__name__)
+
+
+"""Built-in envelopes"""
+
+
+class Envelope(str, Enum):
+    SQS = "sqs"
+    EVENTBRIDGE = "eventbridge"
+    DYNAMODB_STREAM = "dynamodb_stream"
+
+
+class InvalidEnvelopeError(Exception):
+    """Input envelope is not one of the Envelope enum values"""
+
+
+# enum to BaseEnvelope handler class
+__ENVELOPE_MAPPING = {
+    Envelope.SQS: SqsEnvelope,
+    Envelope.DYNAMODB_STREAM: DynamoDBEnvelope,
+    Envelope.EVENTBRIDGE: EventBridgeEnvelope,
+}
+
+
+def parse_envelope(event: Dict[str, Any], envelope: Envelope, schema: BaseModel):
+    envelope_handler: BaseEnvelope = __ENVELOPE_MAPPING.get(envelope)
+    if envelope_handler is None:
+        logger.exception("envelope must be an instance of Envelope enum")
+        raise InvalidEnvelopeError("envelope must be an instance of Envelope enum")
+    logger.debug(f"Parsing and validating event schema, envelope={str(envelope.value)}")
+    return envelope_handler().parse(event=event, schema=schema)

aws_lambda_powertools/utilities/advanced_parser/envelopes/event_bridge.py

@@ -10,10 +10,10 @@
 
 
 class EventBridgeEnvelope(BaseEnvelope):
-    def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+    def parse(self, event: Dict[str, Any], schema: BaseModel) -> Any:
         try:
             parsed_envelope = EventBridgeSchema(**event)
         except (ValidationError, TypeError):
             logger.exception("Validation exception received from input eventbridge event")
             raise
-        return self._parse_user_dict_schema(parsed_envelope.detail, inbound_schema_model)
+        return self._parse_user_dict_schema(parsed_envelope.detail, schema)

aws_lambda_powertools/utilities/advanced_parser/envelopes/sns.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 from pydantic import BaseModel, ValidationError
 
@@ -10,14 +10,13 @@
 
 
 class SqsEnvelope(BaseEnvelope):
-    def parse(self, event: Dict[str, Any], inbound_schema_model: BaseModel) -> Any:
+    def parse(self, event: Dict[str, Any], schema: BaseModel) -> List[BaseModel]:
         try:
             parsed_envelope = SqsSchema(**event)
         except (ValidationError, TypeError):
             logger.exception("Validation exception received from input sqs event")
             raise
         output = []
         for record in parsed_envelope.Records:
-            parsed_msg = self._parse_user_json_string_schema(record.body, inbound_schema_model)
-            output.append({"body": parsed_msg, "attributes": record.messageAttributes})
+            output.append(self._parse_user_json_string_schema(record.body, schema))
         return output

aws_lambda_powertools/utilities/advanced_parser/envelopes/sqs.py

@@ -0,0 +1,56 @@
+import logging
+from typing import Any, Callable, Dict, Optional
+
+from pydantic import BaseModel, ValidationError
+
+from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
+from aws_lambda_powertools.utilities.advanced_parser.envelopes import Envelope, parse_envelope
+
+logger = logging.getLogger(__name__)
+
+
+@lambda_handler_decorator
+def parser(
+    handler: Callable[[Dict, Any], Any],
+    event: Dict[str, Any],
+    context: Dict[str, Any],
+    schema: BaseModel,
+    envelope: Optional[Envelope] = None,
+) -> Any:
+    """Decorator to conduct advanced parsing & validation for lambda handlers events
+
+        As Lambda follows (event, context) signature we can remove some of the boilerplate
+        and also capture any exception any Lambda function throws as metadata.
+        Event will be the parsed & validated BaseModel pydantic object of the input type "schema"
+
+        Example
+        -------
+        **Lambda function using validation decorator**
+
+            @parser(schema=MyBusiness, envelope=envelopes.EVENTBRIDGE)
+            def handler(event: inbound_schema_model , context: LambdaContext):
+                ...
+
+        Parameters
+        ----------
+        todo add
+
+        Raises
+        ------
+        err
+            TypeError or pydantic.ValidationError or any exception raised by the lambda handler itself
+    """
+    lambda_handler_name = handler.__name__
+    parsed_event = None
+    if envelope is None:
+        try:
+            logger.debug("Parsing and validating event schema, no envelope is used")
+            parsed_event = schema(**event)
+        except (ValidationError, TypeError):
+            logger.exception("Validation exception received from input event")
+            raise
+    else:
+        parsed_event = parse_envelope(event, envelope, schema)
+
+    logger.debug(f"Calling handler {lambda_handler_name}")
+    handler(parsed_event, context)

aws_lambda_powertools/utilities/advanced_parser/parser.py

@@ -1,11 +1,12 @@
-from .dynamodb import DynamoDBSchema
+from .dynamodb import DynamoDBSchema, DynamoRecordSchema, DynamoScheme
 from .event_bridge import EventBridgeSchema
-from .sns import SnsSchema
-from .sqs import SqsSchema
+from .sqs import SqsRecordSchema, SqsSchema
 
 __all__ = [
     "DynamoDBSchema",
     "EventBridgeSchema",
-    "SnsSchema",
+    "DynamoScheme",
+    "DynamoRecordSchema",
     "SqsSchema",
+    "SqsRecordSchema",
 ]

aws_lambda_powertools/utilities/advanced_parser/schemas/__init__.py

@@ -6,10 +6,10 @@
 
 
 class DynamoScheme(BaseModel):
-    ApproximateCreationDateTime: date
-    Keys: Dict[Literal["id"], Dict[Literal["S"], str]]
-    NewImage: Optional[Dict[str, Any]] = {}
-    OldImage: Optional[Dict[str, Any]] = {}
+    ApproximateCreationDateTime: Optional[date]
+    Keys: Dict[str, Dict[str, Any]]
+    NewImage: Optional[Dict[str, Any]]
+    OldImage: Optional[Dict[str, Any]]
     SequenceNumber: str
     SizeBytes: int
     StreamViewType: Literal["NEW_AND_OLD_IMAGES", "KEYS_ONLY", "NEW_IMAGE", "OLD_IMAGE"]
@@ -23,6 +23,11 @@ def check_one_image_exists(cls, values):
         return values
 
 
+class UserIdentity(BaseModel):
+    type: Literal["Service"]  # noqa: VNE003, A003
+    principalId: Literal["dynamodb.amazonaws.com"]
+
+
 class DynamoRecordSchema(BaseModel):
     eventID: str
     eventName: Literal["INSERT", "MODIFY", "REMOVE"]
@@ -31,6 +36,7 @@ class DynamoRecordSchema(BaseModel):
     awsRegion: str
     eventSourceARN: str
     dynamodb: DynamoScheme
+    userIdentity: Optional[UserIdentity]
 
 
 class DynamoDBSchema(BaseModel):

aws_lambda_powertools/utilities/advanced_parser/schemas/dynamodb.py

@@ -1,15 +1,16 @@
 from datetime import datetime
 from typing import Any, Dict, List
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 
 class EventBridgeSchema(BaseModel):
     version: str
     id: str  # noqa: A003,VNE003
     source: str
-    account: int
+    account: str
     time: datetime
     region: str
     resources: List[str]
+    detailtype: str = Field(None, alias="detail-type")
     detail: Dict[str, Any]

aws_lambda_powertools/utilities/advanced_parser/schemas/event_bridge.py

@@ -14,6 +14,7 @@ class SqsAttributesSchema(BaseModel):
     SenderId: str
     SentTimestamp: datetime
     SequenceNumber: Optional[str]
+    AWSTraceHeader: Optional[str]
 
 
 class SqsMsgAttributeSchema(BaseModel):
@@ -50,7 +51,7 @@ class SqsRecordSchema(BaseModel):
     attributes: SqsAttributesSchema
     messageAttributes: Dict[str, SqsMsgAttributeSchema]
     md5OfBody: str
-    md5OfMessageAttributes: str
+    md5OfMessageAttributes: Optional[str]
     eventSource: Literal["aws:sqs"]
     eventSourceARN: str
     awsRegion: str