feat(event_sources): add support for VPC Lattice events

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -20,6 +20,7 @@
 from .ses_event import SESEvent
 from .sns_event import SNSEvent
 from .sqs_event import SQSEvent
+from .vpc_lattice import VPCLatticeEvent
 
 __all__ = [
     "APIGatewayProxyEvent",
@@ -42,4 +43,5 @@
     "SNSEvent",
     "SQSEvent",
     "event_source",
+    "VPCLatticeEvent",
 ]

aws_lambda_powertools/utilities/data_classes/vpc_lattice.py

@@ -0,0 +1,91 @@
+import base64
+from typing import Any, Dict, Optional
+
+from aws_lambda_powertools.utilities.data_classes.common import (
+    DictWrapper,
+    get_header_value,
+)
+
+
+class VPCLatticeEvent(DictWrapper):
+    @property
+    def body(self) -> str:
+        """The VPC Lattice body."""
+        return self["body"]
+
+    @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 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"""
+        body: str = self["body"]
+        if self.is_base64_encoded:
+            return base64.b64decode(body.encode()).decode()
+        return body
+
+    @property
+    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"]
+
+    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
+        """
+        params = self.query_string_parameters
+        return default_value if params is None else params.get(name, default_value)
+
+    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(self.headers, name, default_value, case_sensitive)

docs/utilities/data_classes.md

@@ -102,6 +102,7 @@ Log Data Event for Troubleshooting
 | [SES](#ses)                                                               | `SESEvent`                                         |
 | [SNS](#sns)                                                               | `SNSEvent`                                         |
 | [SQS](#sqs)                                                               | `SQSEvent`                                         |
+| [VPC Lattice](#vpc-lattice)                                               | `VPCLatticeEvent`                                  |
 
 ???+ info
     The examples provided below are far from exhaustive - the data classes themselves are designed to provide a form of
@@ -1121,6 +1122,24 @@ This example is based on the AWS Blog post [Introducing Amazon S3 Object Lambda
             do_something_with(record.body)
     ```
 
+### VPC Lattice
+
+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.py"
+    ```
+
+=== "Lattice Example Event"
+
+    ```json
+    --8<-- "examples/event_sources/src/vpc_lattice_payload.json"
+    ```
+
 ## Advanced
 
 ### Debugging

examples/event_sources/src/vpc_lattice.py

@@ -0,0 +1,19 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.data_classes import VPCLatticeEvent, event_source
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+logger = Logger()
+
+
+@event_source(data_class=VPCLatticeEvent)
+def lambda_handler(event: VPCLatticeEvent, context: LambdaContext):
+    logger.info(event.body)
+
+    response = {
+        "isBase64Encoded": False,
+        "statusCode": 200,
+        "headers": {"Content-Type": "application/text"},
+        "body": "Event Response to VPC Lattice 🔥🚀🔥",
+    }
+
+    return response

examples/event_sources/src/vpc_lattice_payload.json

@@ -0,0 +1,15 @@
+{
+    "raw_path": "/testpath",
+    "method": "GET",
+    "headers": {
+      "user_agent": "curl/7.64.1",
+      "x-forwarded-for": "10.213.229.10",
+      "host": "test-lambda-service-3908sdf9u3u.dkfjd93.vpc-lattice-svcs.us-east-2.on.aws",
+      "accept": "*/*"
+    },
+    "query_string_parameters": {
+      "order-id": "1"
+    },
+    "body": "eyJ0ZXN0IjogImV2ZW50In0=",
+    "is_base64_encoded": true
+  }

tests/events/vpcLatticeEvent.json

@@ -0,0 +1,15 @@
+{
+  "raw_path": "/testpath",
+  "method": "GET",
+  "headers": {
+    "user_agent": "curl/7.64.1",
+    "x-forwarded-for": "10.213.229.10",
+    "host": "test-lambda-service-3908sdf9u3u.dkfjd93.vpc-lattice-svcs.us-east-2.on.aws",
+    "accept": "*/*"
+  },
+  "query_string_parameters": {
+    "order-id": "1"
+  },
+  "body": "eyJ0ZXN0IjogImV2ZW50In0=",
+  "is_base64_encoded": true
+}

tests/functional/test_data_classes.py

@@ -25,6 +25,7 @@
     SESEvent,
     SNSEvent,
     SQSEvent,
+    VPCLatticeEvent,
 )
 from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import (
     APIGatewayAuthorizerEventV2,
@@ -2041,3 +2042,18 @@ def test_api_gateway_route_arn_parser():
     details = parse_api_gateway_arn(arn)
     assert details.resource == ""
     assert details.arn == arn + "/"
+
+
+def test_vpc_lattice_event():
+    event = VPCLatticeEvent(load_event("vpcLatticeEvent.json"))
+
+    assert event.raw_path == event["raw_path"]
+    assert event.get_query_string_value("order-id") == "1"
+    assert event.get_header_value("user_agent") == "curl/7.64.1"
+    assert event.decoded_body == '{"test": "event"}'
+    assert event.json_body == {"test": "event"}
+    assert event.method == event["method"]
+    assert event.headers == event["headers"]
+    assert event.query_string_parameters == event["query_string_parameters"]
+    assert event.body == event["body"]
+    assert event.is_base64_encoded == event["is_base64_encoded"]