docs: new parser utility

aws_lambda_powertools/utilities/parser/__init__.py

@@ -2,16 +2,17 @@
 """
 from . import envelopes
 from .envelopes import BaseEnvelope
-from .exceptions import ModelValidationError
-from .parser import event_parser
-from .pydantic import BaseModel, root_validator, validator
+from .parser import event_parser, parse
+from .pydantic import BaseModel, Field, ValidationError, root_validator, validator
 
 __all__ = [
     "event_parser",
+    "parse",
     "envelopes",
     "BaseEnvelope",
     "BaseModel",
+    "Field",
     "validator",
     "root_validator",
-    "ModelValidationError",
+    "ValidationError",
 ]

aws_lambda_powertools/utilities/parser/envelopes/__init__.py

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

aws_lambda_powertools/utilities/parser/envelopes/base.py

@@ -1,8 +1,8 @@
 import logging
 from abc import ABC, abstractmethod
-from typing import Any, Dict, Union
+from typing import Any, Dict, Optional, TypeVar, Union
 
-from pydantic import BaseModel
+from ..types import Model
 
 logger = logging.getLogger(__name__)
 
@@ -11,14 +11,14 @@ class BaseEnvelope(ABC):
     """ABC implementation for creating a supported Envelope"""
 
     @staticmethod
-    def _parse(data: Union[Dict[str, Any], str], model: BaseModel) -> Any:
+    def _parse(data: Optional[Union[Dict[str, Any], Any]], model: Model) -> Union[Model, None]:
         """Parses envelope data against model provided
 
         Parameters
         ----------
         data : Dict
             Data to be parsed and validated
-        model
+        model : Model
             Data model to parse and validate data against
 
         Returns
@@ -38,7 +38,7 @@ def _parse(data: Union[Dict[str, Any], str], model: BaseModel) -> Any:
         return model.parse_obj(data)
 
     @abstractmethod
-    def parse(self, data: Dict[str, Any], model: BaseModel):
+    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Model):
         """Implementation to parse data against envelope model, then against the data model
 
         NOTE: Call `_parse` method to fully parse data with model provided.
@@ -56,3 +56,8 @@ def parse(...):
             return self._parse(data=parsed_envelope.detail, model=data_model)
         """
         return NotImplemented  # pragma: no cover
+
+
+# Generic to support type annotations throughout parser
+# Note: Can't be defined under types.py due to circular dependency
+Envelope = TypeVar("Envelope", bound=BaseEnvelope)

aws_lambda_powertools/utilities/parser/envelopes/dynamodb.py

@@ -1,45 +1,44 @@
 import logging
-from typing import Any, Dict, List
-
-from pydantic import BaseModel
-from typing_extensions import Literal
+from typing import Any, Dict, List, Optional, Union
 
 from ..models import DynamoDBStreamModel
+from ..types import Model
 from .base import BaseEnvelope
 
 logger = logging.getLogger(__name__)
 
 
-class DynamoDBEnvelope(BaseEnvelope):
+class DynamoDBStreamEnvelope(BaseEnvelope):
     """ DynamoDB Stream Envelope to extract data within NewImage/OldImage
 
     Note: Values are the parsed 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], model: BaseModel) -> List[Dict[Literal["NewImage", "OldImage"], BaseModel]]:
+    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Model) -> List[Dict[str, Optional[Model]]]:
         """Parses DynamoDB Stream records found in either NewImage and OldImage with model provided
 
         Parameters
         ----------
         data : Dict
             Lambda event to be parsed
-        model : BaseModel
+        model : Model
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
         List
-            List of records parsed with model provided
+            List of dictionaries with NewImage and OldImage records parsed with model provided
         """
-        parsed_envelope = DynamoDBStreamModel(**data)
+        logger.debug(f"Parsing incoming data with DynamoDB Stream model {DynamoDBStreamModel}")
+        parsed_envelope = DynamoDBStreamModel.parse_obj(data)
         output = []
+        logger.debug(f"Parsing DynamoDB Stream new and old records with {model}")
         for record in parsed_envelope.Records:
             output.append(
                 {
-                    "NewImage": self._parse(record.dynamodb.NewImage, model),
-                    "OldImage": self._parse(record.dynamodb.OldImage, model),
+                    "NewImage": self._parse(data=record.dynamodb.NewImage, model=model),
+                    "OldImage": self._parse(data=record.dynamodb.OldImage, model=model),
                 }
             )
-        # noinspection PyTypeChecker
         return output

aws_lambda_powertools/utilities/parser/envelopes/event_bridge.py

@@ -1,9 +1,8 @@
 import logging
-from typing import Any, Dict
-
-from pydantic import BaseModel
+from typing import Any, Dict, Optional, Union
 
 from ..models import EventBridgeModel
+from ..types import Model
 from .base import BaseEnvelope
 
 logger = logging.getLogger(__name__)
@@ -12,20 +11,22 @@
 class EventBridgeEnvelope(BaseEnvelope):
     """EventBridge envelope to extract data within detail key"""
 
-    def parse(self, data: Dict[str, Any], model: BaseModel) -> BaseModel:
+    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Model) -> Optional[Model]:
         """Parses data found with model provided
 
         Parameters
         ----------
         data : Dict
             Lambda event to be parsed
-        model : BaseModel
+        model : Model
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
         Any
             Parsed detail payload with model provided
         """
-        parsed_envelope = EventBridgeModel(**data)
+        logger.debug(f"Parsing incoming data with EventBridge model {EventBridgeModel}")
+        parsed_envelope = EventBridgeModel.parse_obj(data)
+        logger.debug(f"Parsing event payload in `detail` with {model}")
         return self._parse(data=parsed_envelope.detail, model=model)

aws_lambda_powertools/utilities/parser/envelopes/sqs.py

@@ -1,9 +1,8 @@
 import logging
-from typing import Any, Dict, List, Union
-
-from pydantic import BaseModel
+from typing import Any, Dict, List, Optional, Union
 
 from ..models import SqsModel
+from ..types import Model
 from .base import BaseEnvelope
 
 logger = logging.getLogger(__name__)
@@ -19,23 +18,25 @@ class SqsEnvelope(BaseEnvelope):
     all items in the list will be parsed as str and npt as JSON (and vice versa)
     """
 
-    def parse(self, data: Dict[str, Any], model: Union[BaseModel, str]) -> List[Union[BaseModel, str]]:
+    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Model) -> List[Optional[Model]]:
         """Parses records found with model provided
 
         Parameters
         ----------
         data : Dict
             Lambda event to be parsed
-        model : BaseModel
+        model : Model
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
         List
             List of records parsed with model provided
         """
-        parsed_envelope = SqsModel(**data)
+        logger.debug(f"Parsing incoming data with SQS model {SqsModel}")
+        parsed_envelope = SqsModel.parse_obj(data)
         output = []
+        logger.debug(f"Parsing SQS records in `body` with {model}")
         for record in parsed_envelope.Records:
-            output.append(self._parse(record.body, model))
+            output.append(self._parse(data=record.body, model=model))
         return output

aws_lambda_powertools/utilities/parser/exceptions.py

@@ -2,9 +2,5 @@ class InvalidEnvelopeError(Exception):
     """Input envelope is not callable and instance of BaseEnvelope"""
 
 
-class ModelValidationError(Exception):
-    """Input data does not conform with model"""
-
-
 class InvalidModelTypeError(Exception):
     """Input data model does not implement BaseModel"""

aws_lambda_powertools/utilities/parser/parser.py

@@ -1,23 +1,22 @@
 import logging
 from typing import Any, Callable, Dict, Optional
 
-from pydantic import BaseModel, ValidationError
-
 from ...middleware_factory import lambda_handler_decorator
 from ..typing import LambdaContext
-from .envelopes.base import BaseEnvelope
-from .exceptions import InvalidEnvelopeError, InvalidModelTypeError, ModelValidationError
+from .envelopes.base import Envelope
+from .exceptions import InvalidEnvelopeError, InvalidModelTypeError
+from .types import Model
 
 logger = logging.getLogger(__name__)
 
 
 @lambda_handler_decorator
 def event_parser(
-    handler: Callable[[Dict, Any], Any],
+    handler: Callable[[Any, LambdaContext], Any],
     event: Dict[str, Any],
     context: LambdaContext,
-    model: BaseModel,
-    envelope: Optional[BaseEnvelope] = None,
+    model: Model,
+    envelope: Optional[Envelope] = None,
 ) -> Any:
     """Lambda handler decorator to parse & validate events using Pydantic models
 
@@ -65,14 +64,14 @@ def handler(event: Order, context: LambdaContext):
         Lambda event to be parsed & validated
     context:  LambdaContext
         Lambda context object
-    model:   BaseModel
+    model:   Model
         Your data model that will replace the event.
-    envelope: BaseEnvelope
+    envelope: Envelope
         Optional envelope to extract the model from
 
     Raises
     ------
-    ModelValidationError
+    ValidationError
         When input event does not conform with model provided
     InvalidModelTypeError
         When model given does not implement BaseModel
@@ -84,7 +83,7 @@ def handler(event: Order, context: LambdaContext):
     return handler(parsed_event, context)
 
 
-def parse(event: Dict[str, Any], model: BaseModel, envelope: Optional[BaseEnvelope] = None) -> Any:
+def parse(event: Dict[str, Any], model: Model, envelope: Optional[Envelope] = None) -> Model:
     """Standalone function to parse & validate events using Pydantic models
 
     Typically used when you need fine-grained control over error handling compared to event_parser decorator.
@@ -94,7 +93,7 @@ def parse(event: Dict[str, Any], model: BaseModel, envelope: Optional[BaseEnvelo
 
     **Lambda handler decorator to parse & validate event**
 
-        from aws_lambda_powertools.utilities.parser.exceptions import ModelValidationError
+        from aws_lambda_powertools.utilities.parser import ValidationError
 
         class Order(BaseModel):
             id: int
@@ -104,7 +103,7 @@ class Order(BaseModel):
         def handler(event: Order, context: LambdaContext):
             try:
                 parse(model=Order)
-            except ModelValidationError:
+            except ValidationError:
                 ...
 
     **Lambda handler decorator to parse & validate event - using built-in envelope**
@@ -117,21 +116,21 @@ class Order(BaseModel):
         def handler(event: Order, context: LambdaContext):
             try:
                 parse(model=Order, envelope=envelopes.EVENTBRIDGE)
-            except ModelValidationError:
+            except ValidationError:
                 ...
 
     Parameters
     ----------
     event:    Dict
         Lambda event to be parsed & validated
-    model:   BaseModel
+    model:   Model
         Your data model that will replace the event
-    envelope: BaseEnvelope
+    envelope: Envelope
         Optional envelope to extract the model from
 
     Raises
     ------
-    ModelValidationError
+    ValidationError
         When input event does not conform with model provided
     InvalidModelTypeError
         When model given does not implement BaseModel
@@ -144,13 +143,12 @@ def handler(event: Order, context: LambdaContext):
             return envelope().parse(data=event, model=model)
         except AttributeError:
             raise InvalidEnvelopeError(f"Envelope must implement BaseEnvelope, envelope={envelope}")
-        except (ValidationError, TypeError) as e:
-            raise ModelValidationError(f"Input event does not conform with model, envelope={envelope}") from e
 
     try:
         logger.debug("Parsing and validating event model; no envelope used")
+        if isinstance(event, str):
+            return model.parse_raw(event)
+
         return model.parse_obj(event)
-    except (ValidationError, TypeError) as e:
-        raise ModelValidationError("Input event does not conform with model") from e
     except AttributeError:
-        raise InvalidModelTypeError("Input model must implement BaseModel")
+        raise InvalidModelTypeError(f"Input model must implement BaseModel, model={model}")

aws_lambda_powertools/utilities/parser/types.py

@@ -0,0 +1,6 @@
+"""Generics and other shared types used across parser"""
+from typing import TypeVar
+
+from pydantic import BaseModel
+
+Model = TypeVar("Model", bound=BaseModel)