diff --git a/aws_lambda_powertools/utilities/parser/envelopes/__init__.py b/aws_lambda_powertools/utilities/parser/envelopes/__init__.py
index 0f985f29d88..cbca982adf7 100644
--- a/aws_lambda_powertools/utilities/parser/envelopes/__init__.py
+++ b/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",
]
diff --git a/aws_lambda_powertools/utilities/parser/envelopes/vpc_lattice.py b/aws_lambda_powertools/utilities/parser/envelopes/vpc_lattice.py
new file mode 100644
index 00000000000..fb95ac05a8d
--- /dev/null
+++ b/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)
diff --git a/aws_lambda_powertools/utilities/parser/models/__init__.py b/aws_lambda_powertools/utilities/parser/models/__init__.py
index 952280a519c..ddc76dc7819 100644
--- a/aws_lambda_powertools/utilities/parser/models/__init__.py
+++ b/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",
]
diff --git a/aws_lambda_powertools/utilities/parser/models/vpc_lattice.py b/aws_lambda_powertools/utilities/parser/models/vpc_lattice.py
new file mode 100644
index 00000000000..8442fc92781
--- /dev/null
+++ b/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]
diff --git a/docs/utilities/parser.md b/docs/utilities/parser.md
index 0e25f9441a4..f482dcb0410 100644
--- a/docs/utilities/parser.md
+++ b/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`.
2. Parses `body` key using your model and returns it. | `Model` |
| **LambdaFunctionUrlEnvelope** | 1. Parses data using `LambdaFunctionUrlModel`.
2. Parses `body` key using your model and returns it. | `Model` |
| **KafkaEnvelope** | 1. Parses data using `KafkaRecordModel`.
2. Parses `value` key using your model and returns it. | `Model` |
+| **VpcLatticeEnvelope** | 1. Parses data using `VpcLatticeModel`.
2. Parses `value` key using your model and returns it. | `Model` |
#### Bringing your own envelope
diff --git a/tests/functional/parser/schemas.py b/tests/functional/parser/schemas.py
index 907deb40aa0..1da0213ff45 100644
--- a/tests/functional/parser/schemas.py
+++ b/tests/functional/parser/schemas.py
@@ -99,3 +99,8 @@ class MyLambdaKafkaBusiness(BaseModel):
class MyKinesisFirehoseBusiness(BaseModel):
Hello: str
+
+
+class MyVpcLatticeBusiness(BaseModel):
+ username: str
+ name: str
diff --git a/tests/unit/parser/test_vpc_lattice.py b/tests/unit/parser/test_vpc_lattice.py
new file mode 100644
index 00000000000..f0476509cea
--- /dev/null
+++ b/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)