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

@@ -4,15 +4,15 @@
     BaseHeadersSerializer,
     HttpApiHeadersSerializer,
 )
-from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent
+from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent, DictWrapper
 from aws_lambda_powertools.utilities.data_classes.shared_functions import (
     base64_decode,
     get_header_value,
     get_query_string_value,
 )
 
 
-class VPCLatticeEvent(BaseProxyEvent):
+class VPCLatticeEventBase(BaseProxyEvent):
     @property
     def body(self) -> str:
         """The VPC Lattice body."""
@@ -30,11 +30,6 @@ def headers(self) -> Dict[str, str]:
         """The VPC Lattice event headers."""
         return self["headers"]
 
-    @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 decoded_body(self) -> str:
         """Dynamically base64 decode body as a str"""
@@ -48,29 +43,16 @@ def method(self) -> str:
         """The VPC Lattice method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
         return self["method"]
 
-    @property
-    def query_string_parameters(self) -> Dict[str, str]:
-        """The request query string parameters."""
-        return self["query_string_parameters"]
-
-    @property
-    def raw_path(self) -> str:
-        """The raw VPC Lattice request path."""
-        return self["raw_path"]
-
-    # VPCLattice event has no path field
-    # Added here for consistency with the BaseProxyEvent class
-    @property
-    def path(self) -> str:
-        return self["raw_path"]
-
-    # VPCLattice event has no http_method field
-    # Added here for consistency with the BaseProxyEvent class
     @property
     def http_method(self) -> str:
         """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
         return self["method"]
 
+    @property
+    def query_string_parameters(self) -> Dict[str, str]:
+        """The request query string parameters."""
+        return self["query_string_parameters"]
+
     def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
         """Get query string value by name
 
@@ -140,3 +122,127 @@ def get_header_value(
     def header_serializer(self) -> BaseHeadersSerializer:
         # When using the VPC Lattice integration, we have multiple HTTP Headers.
         return HttpApiHeadersSerializer()
+
+
+class VPCLatticeEvent(VPCLatticeEventBase):
+    @property
+    def raw_path(self) -> str:
+        """The raw VPC Lattice request path."""
+        return self["raw_path"]
+
+    @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"]
+
+    # VPCLattice event has no path field
+    # Added here for consistency with the BaseProxyEvent class
+    @property
+    def path(self) -> str:
+        return self["raw_path"]
+
+
+class vpcLatticeEventV2Identity(DictWrapper):
+    @property
+    def source_vpc_arn(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext Identity sourceVpcArn"""
+        return self.get("sourceVpcArn")
+
+    @property
+    def get_type(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext Identity type"""
+        return self.get("type")
+
+    @property
+    def principal(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext principal"""
+        return self.get("principal")
+
+    @property
+    def principal_org_id(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext principalOrgID"""
+        return self.get("principalOrgID")
+
+    @property
+    def session_name(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext sessionName"""
+        return self.get("sessionName")
+
+    @property
+    def x509_subject_cn(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext X509SubjectCn"""
+        return self.get("X509SubjectCn")
+
+    @property
+    def x509_issuer_ou(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext X509IssuerOu"""
+        return self.get("X509IssuerOu")
+
+    @property
+    def x509_san_dns(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext X509SanDns"""
+        return self.get("x509SanDns")
+
+    @property
+    def x509_san_uri(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext X509SanUri"""
+        return self.get("X509SanUri")
+
+    @property
+    def x509_san_name_cn(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext X509SanNameCn"""
+        return self.get("X509SanNameCn")
+
+
+class vpcLatticeEventV2RequestContext(DictWrapper):
+    @property
+    def service_network_arn(self) -> str:
+        """The VPC Lattice v2 Event requestContext serviceNetworkArn"""
+        return self["serviceNetworkArn"]
+
+    @property
+    def service_arn(self) -> str:
+        """The VPC Lattice v2 Event requestContext serviceArn"""
+        return self["serviceArn"]
+
+    @property
+    def target_group_arn(self) -> str:
+        """The VPC Lattice v2 Event requestContext targetGroupArn"""
+        return self["targetGroupArn"]
+
+    @property
+    def identity(self) -> vpcLatticeEventV2Identity:
+        """The VPC Lattice v2 Event requestContext identity"""
+        return vpcLatticeEventV2Identity(self["identity"])
+
+    @property
+    def region(self) -> str:
+        """The VPC Lattice v2 Event requestContext serviceNetworkArn"""
+        return self["region"]
+
+    @property
+    def time_epoch(self) -> str:
+        """The VPC Lattice v2 Event requestContext timeEpoch"""
+        return self["timeEpoch"]
+
+
+class VPCLatticeEventV2(VPCLatticeEventBase):
+    @property
+    def version(self) -> str:
+        """The VPC Lattice v2 Event version"""
+        return self["version"]
+
+    @property
+    def is_base64_encoded(self) -> Optional[bool]:
+        """A boolean flag to indicate if the applicable request payload is Base64-encode"""
+        return self.get("isBase64Encoded")
+
+    @property
+    def path(self) -> str:
+        """The VPC Lattice v2 Event path"""
+        return self["path"]
+
+    @property
+    def request_context(self) -> vpcLatticeEventV2RequestContext:
+        """he VPC Lattice v2 Event request context."""
+        return vpcLatticeEventV2RequestContext(self["requestContext"])

aws_lambda_powertools/utilities/parser/envelopes/__init__.py

@@ -11,6 +11,7 @@
 from .sns import SnsEnvelope, SnsSqsEnvelope
 from .sqs import SqsEnvelope
 from .vpc_lattice import VpcLatticeEnvelope
+from .vpc_latticev2 import VpcLatticeV2Envelope
 
 __all__ = [
     "ApiGatewayEnvelope",
@@ -27,4 +28,5 @@
     "KafkaEnvelope",
     "BaseEnvelope",
     "VpcLatticeEnvelope",
+    "VpcLatticeV2Envelope",
 ]

aws_lambda_powertools/utilities/parser/envelopes/vpc_latticev2.py

@@ -0,0 +1,32 @@
+import logging
+from typing import Any, Dict, Optional, Type, Union
+
+from ..models import VpcLatticeV2Model
+from ..types import Model
+from .base import BaseEnvelope
+
+logger = logging.getLogger(__name__)
+
+
+class VpcLatticeV2Envelope(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 V2 model {VpcLatticeV2Model}")
+        parsed_envelope: VpcLatticeV2Model = VpcLatticeV2Model.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

@@ -89,6 +89,7 @@
 from .sns import SnsModel, SnsNotificationModel, SnsRecordModel
 from .sqs import SqsAttributesModel, SqsModel, SqsMsgAttributeModel, SqsRecordModel
 from .vpc_lattice import VpcLatticeModel
+from .vpc_latticev2 import VpcLatticeV2Model
 
 __all__ = [
     "APIGatewayProxyEventV2Model",
@@ -163,4 +164,5 @@
     "CloudFormationCustomResourceCreateModel",
     "CloudFormationCustomResourceBaseModel",
     "VpcLatticeModel",
+    "VpcLatticeV2Model",
 ]

aws_lambda_powertools/utilities/parser/models/vpc_latticev2.py

@@ -0,0 +1,41 @@
+from datetime import datetime
+from typing import Dict, Optional, Type, Union
+
+from pydantic import BaseModel, Field, validator
+
+
+class VpcLatticeV2RequestContextIdentity(BaseModel):
+    source_vpc_arn: Optional[str] = Field(None, alias="sourceVpcArn")
+    get_type: Optional[str] = Field(None, alias="type")
+    principal: Optional[str] = Field(None, alias="principal")
+    principal_org_id: Optional[str] = Field(None, alias="principalOrgID")
+    session_name: Optional[str] = Field(None, alias="sessionName")
+    x509_subject_cn: Optional[str] = Field(None, alias="X509SubjectCn")
+    x509_issuer_ou: Optional[str] = Field(None, alias="X509IssuerOu")
+    x509_san_dns: Optional[str] = Field(None, alias="x509SanDns")
+    x509_san_uri: Optional[str] = Field(None, alias="X509SanUri")
+    x509_san_name_cn: Optional[str] = Field(None, alias="X509SanNameCn")
+
+
+class VpcLatticeV2RequestContext(BaseModel):
+    service_network_arn: str = Field(alias="serviceNetworkArn")
+    service_arn: str = Field(alias="serviceArn")
+    target_group_arn: str = Field(alias="targetGroupArn")
+    identity: VpcLatticeV2RequestContextIdentity
+    region: str
+    time_epoch: datetime = Field(alias="timeEpoch")
+
+    @validator("time_epoch", pre=True, allow_reuse=True)
+    def time_epoch_convert_to_miliseconds(cls, value: int):
+        return round(int(value) / 1000)
+
+
+class VpcLatticeV2Model(BaseModel):
+    version: str
+    path: str
+    method: str
+    headers: Dict[str, str]
+    query_string_parameters: Optional[Dict[str, str]] = None
+    body: Optional[Union[str, Type[BaseModel]]] = None
+    is_base64_encoded: Optional[bool] = Field(None, alias="isBase64Encoded")
+    request_context: VpcLatticeV2RequestContext = Field(None, alias="requestContext")

docs/utilities/data_classes.md

@@ -104,6 +104,7 @@ Log Data Event for Troubleshooting
 | [SNS](#sns)                                                               | `SNSEvent`                                         |
 | [SQS](#sqs)                                                               | `SQSEvent`                                         |
 | [VPC Lattice](#vpc-lattice)                                               | `VPCLatticeEvent`                                  |
+| [VPC Lattice V2](#vpc-lattice-v2)                                         | `VPCLatticeV2Event`                                |
 
 ???+ info
     The examples provided below are far from exhaustive - the data classes themselves are designed to provide a form of
@@ -1198,6 +1199,24 @@ You can register your Lambda functions as targets within an Amazon VPC Lattice s
     --8<-- "examples/event_sources/src/vpc_lattice_payload.json"
     ```
 
+### VPC Lattice V2
+
+You can register your Lambda functions as targets within an Amazon VPC Lattice service network. By doing this, your Lambda function becomes a service within the network, and clients that have access to the VPC Lattice service network can call your service.
+
+[Click here](https://docs.aws.amazon.com/lambda/latest/dg/services-vpc-lattice.html){target="_blank"} for more information about using AWS Lambda with Amazon VPC Lattice.
+
+=== "app.py"
+
+    ```python hl_lines="2 8"
+    --8<-- "examples/event_sources/src/vpc_lattice_v2.py"
+    ```
+
+=== "Lattice Example Event"
+
+    ```json
+    --8<-- "examples/event_sources/src/vpc_lattice_payload_v2.json"
+    ```
+
 ## Advanced
 
 ### Debugging

docs/utilities/parser.md

@@ -192,6 +192,7 @@ Parser comes with the following built-in models:
 | **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                                    |
+| **VpcLatticeV2Model**                       | Lambda Event Source payload for Amazon VPC Lattice v2 payload                         |
 
 #### Extending built-in models
 

examples/event_sources/src/vpc_lattice_v2.py

@@ -0,0 +1,20 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.data_classes import VPCLatticeEventV2, event_source
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+logger = Logger()
+
+
+@event_source(data_class=VPCLatticeEventV2)
+def lambda_handler(event: VPCLatticeEventV2, context: LambdaContext):
+    logger.info(event.body)
+
+    response = {
+        "isBase64Encoded": False,
+        "statusCode": 200,
+        "statusDescription": "200 OK",
+        "headers": {"Content-Type": "application/text"},
+        "body": "VPC Lattice V2 Event ✨🎉✨",
+    }
+
+    return response