chore: ease maintenance of upcoming parser #118

aws_lambda_powertools/utilities/parser/__init__.py

@@ -0,0 +1,9 @@
+"""Advanced parser utility
+"""
+from . import envelopes
+from .envelopes import BaseEnvelope
+from .exceptions import SchemaValidationError
+from .parser import parser
+from .pydantic import BaseModel, root_validator, validator
+
+__all__ = ["parser", "envelopes", "BaseEnvelope", "BaseModel", "validator", "root_validator", "SchemaValidationError"]

aws_lambda_powertools/utilities/parser/envelopes/__init__.py

@@ -0,0 +1,6 @@
+from .base import BaseEnvelope
+from .dynamodb import DynamoDBEnvelope
+from .event_bridge import EventBridgeEnvelope
+from .sqs import SqsEnvelope
+
+__all__ = ["DynamoDBEnvelope", "EventBridgeEnvelope", "SqsEnvelope", "BaseEnvelope"]

aws_lambda_powertools/utilities/parser/envelopes/base.py

@@ -0,0 +1,58 @@
+import logging
+from abc import ABC, abstractmethod
+from typing import Any, Dict, Union
+
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+
+
+class BaseEnvelope(ABC):
+    """ABC implementation for creating a supported Envelope"""
+
+    @staticmethod
+    def _parse(data: Union[Dict[str, Any], str], schema: BaseModel) -> Any:
+        """Parses envelope data against schema provided
+
+        Parameters
+        ----------
+        data : Dict
+            Data to be parsed and validated
+        schema
+            Schema to parse and validate data against
+
+        Returns
+        -------
+        Any
+            Parsed data
+        """
+        if data is None:
+            logger.debug("Skipping parsing as event is None")
+            return data
+
+        logger.debug("parsing event against schema")
+        if isinstance(data, str):
+            logger.debug("parsing event as string")
+            return schema.parse_raw(data)
+
+        return schema.parse_obj(data)
+
+    @abstractmethod
+    def parse(self, data: Dict[str, Any], schema: BaseModel):
+        """Implementation to parse data against envelope schema, then against the schema
+
+        NOTE: Call `_parse` method to fully parse data with schema provided.
+
+        Example
+        -------
+
+        **EventBridge envelope implementation example**
+
+        def parse(...):
+            # 1. parses data against envelope schema
+            parsed_envelope = EventBridgeSchema(**data)
+
+            # 2. parses portion of data within the envelope against schema
+            return self._parse(data=parsed_envelope.detail, schema=schema)
+        """
+        return NotImplemented  # pragma: no cover

aws_lambda_powertools/utilities/parser/envelopes/dynamodb.py

@@ -0,0 +1,45 @@
+import logging
+from typing import Any, Dict, List
+
+from pydantic import BaseModel
+from typing_extensions import Literal
+
+from ..schemas import DynamoDBStreamSchema
+from .base import BaseEnvelope
+
+logger = logging.getLogger(__name__)
+
+
+class DynamoDBEnvelope(BaseEnvelope):
+    """ DynamoDB Stream Envelope to extract data within NewImage/OldImage
+
+    Note: Values are the parsed schema models. Images' values can also be None, and
+    length of the list is the record's amount in the original event.
+    """
+
+    def parse(self, data: Dict[str, Any], schema: BaseModel) -> List[Dict[Literal["NewImage", "OldImage"], BaseModel]]:
+        """Parses DynamoDB Stream records found in either NewImage and OldImage with schema provided
+
+        Parameters
+        ----------
+        data : Dict
+            Lambda event to be parsed
+        schema : BaseModel
+            User schema provided to parse after extracting data using envelope
+
+        Returns
+        -------
+        List
+            List of records parsed with schema provided
+        """
+        parsed_envelope = DynamoDBStreamSchema(**data)
+        output = []
+        for record in parsed_envelope.Records:
+            output.append(
+                {
+                    "NewImage": self._parse(record.dynamodb.NewImage, schema),
+                    "OldImage": self._parse(record.dynamodb.OldImage, schema),
+                }
+            )
+        # noinspection PyTypeChecker
+        return output