improv: address Koudai's PR feedback

Author: heitorlessa

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, 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: Union[Dict[str, Any], str], model: Model) -> Model:
         """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: Dict[str, 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,10 +1,10 @@
 import logging
 from typing import Any, Dict, List
 
-from pydantic import BaseModel
 from typing_extensions import Literal
 
 from ..models import DynamoDBStreamModel
+from ..types import Model
 from .base import BaseEnvelope
 
 logger = logging.getLogger(__name__)
@@ -17,14 +17,14 @@ class DynamoDBStreamEnvelope(BaseEnvelope):
     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: Dict[str, Any], model: Model) -> List[Dict[Literal["NewImage", "OldImage"], 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
@@ -33,14 +33,14 @@ def parse(self, data: Dict[str, Any], model: BaseModel) -> List[Dict[Literal["Ne
             List of records parsed with model provided
         """
         logger.debug(f"Parsing incoming data with DynamoDB Stream model {DynamoDBStreamModel}")
-        parsed_envelope = DynamoDBStreamModel(**data)
+        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

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 ..models import EventBridgeModel
+from ..types import Model
 from .base import BaseEnvelope
 
 logger = logging.getLogger(__name__)
@@ -12,14 +11,14 @@
 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: Dict[str, Any], model: Model) -> 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
@@ -28,6 +27,6 @@ def parse(self, data: Dict[str, Any], model: BaseModel) -> BaseModel:
             Parsed detail payload with model provided
         """
         logger.debug(f"Parsing incoming data with EventBridge model {EventBridgeModel}")
-        parsed_envelope = EventBridgeModel(**data)
+        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
 
 from ..models import SqsModel
+from ..types import Model
 from .base import BaseEnvelope
 
 logger = logging.getLogger(__name__)
@@ -19,14 +18,14 @@ 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[BaseModel]:
+    def parse(self, data: Dict[str, Any], model: Model) -> List[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
@@ -35,9 +34,9 @@ def parse(self, data: Dict[str, Any], model: Union[BaseModel, str]) -> List[Base
             List of records parsed with model provided
         """
         logger.debug(f"Parsing incoming data with SQS model {SqsModel}")
-        parsed_envelope = SqsModel(**data)
+        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/parser.py

@@ -1,15 +1,12 @@
 import logging
-from typing import Any, Callable, Dict, Optional, Type, TypeVar
-
-from pydantic import BaseModel
+from typing import Any, Callable, Dict, Optional
 
 from ...middleware_factory import lambda_handler_decorator
 from ..typing import LambdaContext
-from .envelopes.base import BaseEnvelope
+from .envelopes.base import Envelope
 from .exceptions import InvalidEnvelopeError, InvalidModelTypeError
+from .types import Model
 
-Model = TypeVar("Model", bound=BaseModel)
-Envelope = TypeVar("Envelope", bound=BaseEnvelope)
 logger = logging.getLogger(__name__)
 
 
@@ -18,8 +15,8 @@ def event_parser(
     handler: Callable[[Dict, Any], Any],
     event: Dict[str, Any],
     context: LambdaContext,
-    model: Type[Model],
-    envelope: Optional[Type[Envelope]] = None,
+    model: Model,
+    envelope: Optional[Envelope] = None,
 ) -> Any:
     """Lambda handler decorator to parse & validate events using Pydantic models
 
@@ -67,9 +64,9 @@ 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
@@ -86,7 +83,7 @@ def handler(event: Order, context: LambdaContext):
     return handler(parsed_event, context)
 
 
-def parse(event: Dict[str, Any], model: Type[Model], envelope: Optional[Type[Envelope]] = 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.
@@ -126,9 +123,9 @@ def handler(event: Order, context: LambdaContext):
     ----------
     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

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)