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,36 @@
+from typing import Dict
+
+from aws_lambda_powertools.utilities.data_classes.common import (
+    DictWrapper,
+)
+
+class VPCLatticeEvent(DictWrapper):
+    @property
+    def body(self) -> str:
+        """The VPC Lattice body."""
+        return self["body"]
+
+    @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 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"]

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,71 @@ This example is based on the AWS Blog post [Introducing Amazon S3 Object Lambda
             do_something_with(record.body)
     ```
 
+### VPC Lattice
+
+The VPC Lattice service supports Lambda invocation for requests over both HTTP and HTTPS. The service sends an event in JSON format, and adds the `X-Forwarded-For` header to every request. The service also adds the X-Forwarded-Proto header to requests over HTTPS.
+
+The service Base64 encodes the body and sets isBase64Encoded to true, if the content-encoding header is present, and the content type is not one of the following:
+
+* `text/*`
+* `application/json`
+* `application/xml`
+* `application/javascript`
+
+If the content-encoding header is not present, Base64 encoding depends on the content type. For the content types, `text/*`, `application/json`, `application/xml`, and `application/javascript`, the service sends the body as is and sets `is_base64_encoded` to `false`.
+
+=== "app.py"
+
+    ```python
+    from aws_lambda_powertools.utilities.data_classes import event_source, VPCLatticeEvent
+
+    @event_source(data_class=VPCLatticeEvent)
+    def lambda_handler(event: VPCLatticeEvent, context):
+        do_something_with(event.body)
+
+    ```
+
+=== "Lattice Example Event"
+
+    ```json
+    {
+        "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": {
+            "key": "value"
+        },
+        "body": "...",
+        "is_base64_encoded": false
+    }
+    ```
+
+=== "Lattice Return Object"
+
+    ```python
+    response = {
+        "isBase64Encoded": False,
+        "statusCode": 200,
+        headers: {
+            "Content-Type": "application/text"
+        },
+        body: "Event Response to VPC Lattice 🔥🚀🔥"
+    }
+
+    return response
+    ```
+
+The response from your Lambda function must include the Base64 encoding status, status code, and headers. You can omit the body.
+
+To include a binary content in the body of the response, you must Base64 encode the content and set `isBase64Encoded` to `True`. The service decodes the content to retrieve the binary content and sends it to the client in the body of the HTTP response.
+
+The VPC Lattice service does not honor hop-by-hop headers, such as `Connection` or `Transfer-Encoding`. You can omit the `Content-Length` header because the service computes it before sending responses to clients.
+
 ## Advanced
 
 ### Debugging

tests/events/vpcLatticeEvent.json

@@ -0,0 +1,13 @@
+{
+  "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": {},
+  "body": "",
+  "is_base64_encoded": false
+}

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,14 @@ 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.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"]