feat(parser): add support to VpcLatticeModel

aws_lambda_powertools/utilities/parser/envelopes/__init__.py

@@ -10,6 +10,7 @@
 from .lambda_function_url import LambdaFunctionUrlEnvelope
 from .sns import SnsEnvelope, SnsSqsEnvelope
 from .sqs import SqsEnvelope
+from .vpc_lattice import VpcLatticeEnvelope
 
 __all__ = [
     "ApiGatewayEnvelope",
@@ -25,4 +26,5 @@
     "SqsEnvelope",
     "KafkaEnvelope",
     "BaseEnvelope",
+    "VpcLatticeEnvelope",
 ]

aws_lambda_powertools/utilities/parser/envelopes/vpc_lattice.py

@@ -0,0 +1,32 @@
+import logging
+from typing import Any, Dict, Optional, Type, Union
+
+from ..models import VpcLatticeModel
+from ..types import Model
+from .base import BaseEnvelope
+
+logger = logging.getLogger(__name__)
+
+
+class VpcLatticeEnvelope(BaseEnvelope):
+    """Amazon VPC Lattice envelope to extract data within body key"""
+
+    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]:
+        """Parses data found with model provided
+
+        Parameters
+        ----------
+        data : Dict
+            Lambda event to be parsed
+        model : Type[Model]
+            Data model provided to parse after extracting data using envelope
+
+        Returns
+        -------
+        Optional[Model]
+            Parsed detail payload with model provided
+        """
+        logger.debug(f"Parsing incoming data with VPC Lattice model {VpcLatticeModel}")
+        parsed_envelope: VpcLatticeModel = VpcLatticeModel.parse_obj(data)
+        logger.debug(f"Parsing event payload in `detail` with {model}")
+        return self._parse(data=parsed_envelope.body, model=model)

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -84,6 +84,7 @@
 )
 from .sns import SnsModel, SnsNotificationModel, SnsRecordModel
 from .sqs import SqsAttributesModel, SqsModel, SqsMsgAttributeModel, SqsRecordModel
+from .vpc_lattice import VpcLatticeModel
 
 __all__ = [
     "APIGatewayProxyEventV2Model",
@@ -157,4 +158,5 @@
     "CloudFormationCustomResourceDeleteModel",
     "CloudFormationCustomResourceCreateModel",
     "CloudFormationCustomResourceBaseModel",
+    "VpcLatticeModel",
 ]

aws_lambda_powertools/utilities/parser/models/vpc_lattice.py

@@ -0,0 +1,12 @@
+from typing import Dict, Type, Union
+
+from pydantic import BaseModel
+
+
+class VpcLatticeModel(BaseModel):
+    method: str
+    raw_path: str
+    body: Union[str, Type[BaseModel]]
+    is_base64_encoded: bool
+    headers: Dict[str, str]
+    query_string_parameters: Dict[str, str]

docs/utilities/parser.md

@@ -180,6 +180,7 @@ Parser comes with the following built-in models:
 | **SesModel**                                | Lambda Event Source payload for Amazon Simple Email Service                           |
 | **SnsModel**                                | Lambda Event Source payload for Amazon Simple Notification Service                    |
 | **SqsModel**                                | Lambda Event Source payload for Amazon SQS                                            |
+| **VpcLatticeModel**                         | Lambda Event Source payload for Amazon VPC Lattice                                    |
 
 #### Extending built-in models
 
@@ -336,6 +337,7 @@ Parser comes with the following built-in envelopes, where `Model` in the return
 | **ApiGatewayV2Envelope**      | 1. Parses data using `APIGatewayProxyEventV2Model`. <br/> 2. Parses `body` key using your model and returns it.                                                                                             | `Model`                            |
 | **LambdaFunctionUrlEnvelope** | 1. Parses data using `LambdaFunctionUrlModel`. <br/> 2. Parses `body` key using your model and returns it.                                                                                                  | `Model`                            |
 | **KafkaEnvelope**             | 1. Parses data using `KafkaRecordModel`. <br/> 2. Parses `value` key using your model and returns it.                                                                                                       | `Model`                            |
+| **VpcLatticeEnvelope**        | 1. Parses data using `VpcLatticeModel`. <br/> 2. Parses `value` key using your model and returns it.                                                                                                       | `Model`                            |
 
 #### Bringing your own envelope
 

tests/functional/parser/schemas.py

@@ -99,3 +99,8 @@ class MyLambdaKafkaBusiness(BaseModel):
 
 class MyKinesisFirehoseBusiness(BaseModel):
     Hello: str
+
+
+class MyVpcLatticeBusiness(BaseModel):
+    username: str
+    name: str

tests/unit/parser/test_vpc_lattice.py

@@ -0,0 +1,53 @@
+import pytest
+
+from aws_lambda_powertools.utilities.parser import (
+    ValidationError,
+    envelopes,
+    event_parser,
+)
+from aws_lambda_powertools.utilities.parser.models import VpcLatticeModel
+from aws_lambda_powertools.utilities.typing import LambdaContext
+from tests.functional.parser.schemas import MyVpcLatticeBusiness
+from tests.functional.utils import load_event
+
+
+@event_parser(model=MyVpcLatticeBusiness, envelope=envelopes.VpcLatticeEnvelope)
+def handle_lambda_vpclattice_with_envelope(event: MyVpcLatticeBusiness, context: LambdaContext):
+    assert event.username == "Leandro"
+    assert event.name == "Damascena"
+
+
+def test_vpc_lattice_event_with_envelope():
+    event = load_event("vpcLatticeEvent.json")
+    event["body"] = '{"username": "Leandro", "name": "Damascena"}'
+    handle_lambda_vpclattice_with_envelope(event, LambdaContext())
+
+
+def test_vpc_lattice_event():
+    raw_event = load_event("vpcLatticeEvent.json")
+    model = VpcLatticeModel(**raw_event)
+
+    assert model.body == raw_event["body"]
+    assert model.method == raw_event["method"]
+    assert model.raw_path == raw_event["raw_path"]
+    assert model.is_base64_encoded == raw_event["is_base64_encoded"]
+    assert model.headers == raw_event["headers"]
+    assert model.query_string_parameters == raw_event["query_string_parameters"]
+
+
+def test_vpc_lattice_event_custom_model():
+    class MyCustomResource(VpcLatticeModel):
+        body: str
+
+    raw_event = load_event("vpcLatticeEvent.json")
+    model = MyCustomResource(**raw_event)
+
+    assert model.body == raw_event["body"]
+
+
+def test_vpc_lattice_event_invalid():
+    raw_event = load_event("vpcLatticeEvent.json")
+    raw_event["body"] = ["some_data"]
+
+    with pytest.raises(ValidationError):
+        VpcLatticeModel(**raw_event)