feat(event_handler): add support to VPC Lattice payload v2

aws_lambda_powertools/event_handler/api_gateway.py

@@ -22,6 +22,7 @@
     APIGatewayProxyEventV2,
     LambdaFunctionUrlEvent,
     VPCLatticeEvent,
+    VPCLatticeEventV2
 )
 from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent
 from aws_lambda_powertools.utilities.typing import LambdaContext
@@ -43,6 +44,7 @@ class ProxyEventType(Enum):
     APIGatewayProxyEventV2 = "APIGatewayProxyEventV2"
     ALBEvent = "ALBEvent"
     VPCLatticeEvent = "VPCLatticeEvent"
+    VPCLatticeEventV2 = "VPCLatticeEventV2"
     LambdaFunctionUrlEvent = "LambdaFunctionUrlEvent"
 
 
@@ -999,6 +1001,9 @@ def _to_proxy_event(self, event: Dict) -> BaseProxyEvent:
         if self._proxy_type == ProxyEventType.VPCLatticeEvent:
             logger.debug("Converting event to VPC Lattice contract")
             return VPCLatticeEvent(event)
+        if self._proxy_type == ProxyEventType.VPCLatticeEventV2:
+            logger.debug("Converting event to VPC Lattice contract")
+            return VPCLatticeEventV2(event)
         logger.debug("Converting event to ALB contract")
         return ALBEvent(event)
 

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -27,7 +27,7 @@
 from .ses_event import SESEvent
 from .sns_event import SNSEvent
 from .sqs_event import SQSEvent
-from .vpc_lattice import VPCLatticeEvent
+from .vpc_lattice import VPCLatticeEvent, VPCLatticeEventV2
 
 __all__ = [
     "APIGatewayProxyEvent",
@@ -56,4 +56,5 @@
     "event_source",
     "AWSConfigRuleEvent",
     "VPCLatticeEvent",
+    "VPCLatticeEventV2",
 ]

aws_lambda_powertools/utilities/data_classes/vpc_lattice.py

@@ -140,3 +140,130 @@ def get_header_value(
     def header_serializer(self) -> BaseHeadersSerializer:
         # When using the VPC Lattice integration, we have multiple HTTP Headers.
         return HttpApiHeadersSerializer()
+
+
+class VPCLatticeEventV2(BaseProxyEvent):
+    @property
+    def version(self) -> str:
+        """The VPC Lattice v2 Event version"""
+        return self["version"]
+
+    @property
+    def path(self) -> str:
+        """The VPC Lattice v2 Event path"""
+        return self["path"]
+
+    @property
+    def method(self) -> str:
+        """The VPC Lattice v2 Event method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
+        return self["method"]
+
+    @property
+    def headers(self) -> Dict[str, str]:
+        """The VPC Lattice v2 Event headers."""
+        return self["headers"]
+
+    @property
+    def query_string_parameters(self) -> Dict[str, str]:
+        """The request query string parameters."""
+        return self["query_string_parameters"]
+
+    @property
+    def body(self) -> str:
+        """The VPC Lattice v2 Event body."""
+        return self["body"]
+
+    @property
+    def is_base64_encoded(self) -> bool:
+        """A boolean flag to indicate if the applicable request payload is Base64-encode"""
+        return self["is_base64_encoded"]
+
+    @property
+    def request_context(self) -> Dict[str, str]:
+        """he VPC Lattice v2 Event request context."""
+        return self["requestContext"]
+
+    @property
+    def json_body(self) -> Any:
+        """Parses the submitted body as json"""
+        if self._json_data is None:
+            self._json_data = self._json_deserializer(self.decoded_body)
+        return self._json_data
+
+    @property
+    def decoded_body(self) -> str:
+        """Dynamically base64 decode body as a str"""
+        body: str = self["body"]
+        if self.is_base64_encoded:
+            return base64_decode(body)
+        return body
+
+    def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
+        """Get query string value by name
+
+        Parameters
+        ----------
+        name: str
+            Query string parameter name
+        default_value: str, optional
+            Default value if no value was found by name
+        Returns
+        -------
+        str, optional
+            Query string parameter value
+        """
+        return get_query_string_value(
+            query_string_parameters=self.query_string_parameters,
+            name=name,
+            default_value=default_value,
+        )
+
+    @overload
+    def get_header_value(
+        self,
+        name: str,
+        default_value: str,
+        case_sensitive: Optional[bool] = False,
+    ) -> str:
+        ...
+
+    @overload
+    def get_header_value(
+        self,
+        name: str,
+        default_value: Optional[str] = None,
+        case_sensitive: Optional[bool] = False,
+    ) -> Optional[str]:
+        ...
+
+    def get_header_value(
+        self,
+        name: str,
+        default_value: Optional[str] = None,
+        case_sensitive: Optional[bool] = False,
+    ) -> Optional[str]:
+        """Get header value by name
+
+        Parameters
+        ----------
+        name: str
+            Header name
+        default_value: str, optional
+            Default value if no value was found by name
+        case_sensitive: bool
+            Whether to use a case-sensitive look up
+        Returns
+        -------
+        str, optional
+            Header value
+        """
+        return get_header_value(
+            headers=self.headers,
+            name=name,
+            default_value=default_value,
+            case_sensitive=case_sensitive,
+        )
+
+    def header_serializer(self) -> BaseHeadersSerializer:
+        # When using the VPC Lattice integration, we have multiple HTTP Headers.
+        return HttpApiHeadersSerializer()