feat(trigger): Application load balancer event

Create BaseProxyEvent for some reusable code for the http proxy events
Add support for ALB events

Author: Michael Brewer

aws_lambda_powertools/utilities/trigger/__init__.py

@@ -1,3 +1,4 @@
+from .alb_event import ALBEvent
 from .api_gateway_proxy_event import APIGatewayProxyEvent, APIGatewayProxyEventV2
 from .cloud_watch_logs_event import CloudWatchLogsEvent
 from .dynamo_db_stream_event import DynamoDBStreamEvent
@@ -11,6 +12,7 @@
 __all__ = [
     "APIGatewayProxyEvent",
     "APIGatewayProxyEventV2",
+    "ALBEvent",
     "CloudWatchLogsEvent",
     "DynamoDBStreamEvent",
     "EventBridgeEvent",

aws_lambda_powertools/utilities/trigger/alb_event.py

@@ -0,0 +1,38 @@
+from typing import Dict, List, Optional
+
+from aws_lambda_powertools.utilities.trigger.common import BaseProxyEvent, DictWrapper
+
+
+class ALBEventRequestContext(DictWrapper):
+    @property
+    def elb_target_group_arn(self) -> str:
+        return self["requestContext"]["elb"]["targetGroupArn"]
+
+
+class ALBEvent(BaseProxyEvent):
+    """Application load balancer event
+
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/lambda/latest/dg/services-alb.html
+    """
+
+    @property
+    def request_context(self) -> ALBEventRequestContext:
+        return ALBEventRequestContext(self)
+
+    @property
+    def http_method(self) -> str:
+        return self["httpMethod"]
+
+    @property
+    def path(self) -> str:
+        return self["path"]
+
+    @property
+    def multi_value_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
+        return self.get("multiValueQueryStringParameters")
+
+    @property
+    def multi_value_headers(self) -> Optional[Dict[str, List[str]]]:
+        return self.get("multiValueHeaders")

aws_lambda_powertools/utilities/trigger/api_gateway_proxy_event.py

@@ -1,6 +1,6 @@
 from typing import Any, Dict, List, Optional
 
-from aws_lambda_powertools.utilities.trigger.common import DictWrapper
+from aws_lambda_powertools.utilities.trigger.common import BaseProxyEvent, DictWrapper
 
 
 class APIGatewayEventIdentity(DictWrapper):
@@ -196,7 +196,7 @@ def route_key(self) -> Optional[str]:
         return self["requestContext"].get("routeKey")
 
 
-class APIGatewayProxyEvent(dict):
+class APIGatewayProxyEvent(BaseProxyEvent):
     """AWS Lambda proxy V1
 
     Documentation:
@@ -221,18 +221,10 @@ def http_method(self) -> str:
         """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
         return self["httpMethod"]
 
-    @property
-    def headers(self) -> Dict[str, str]:
-        return self["headers"]
-
     @property
     def multi_value_headers(self) -> Dict[str, List[str]]:
         return self["multiValueHeaders"]
 
-    @property
-    def query_string_parameters(self) -> Optional[Dict[str, str]]:
-        return self.get("queryStringParameters")
-
     @property
     def multi_value_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
         return self.get("multiValueQueryStringParameters")
@@ -249,47 +241,6 @@ def path_parameters(self) -> Optional[Dict[str, str]]:
     def stage_variables(self) -> Optional[Dict[str, str]]:
         return self.get("stageVariables")
 
-    @property
-    def body(self) -> Optional[str]:
-        return self.get("body")
-
-    @property
-    def is_base64_encoded(self) -> bool:
-        return self["isBase64Encoded"]
-
-    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) -> 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
-        Returns
-        -------
-        str, optional
-            Header value
-        """
-        return self.headers.get(name, default_value)
-
 
 class RequestContextV2Http(DictWrapper):
     @property
@@ -381,7 +332,7 @@ def time_epoch(self) -> int:
         return self["requestContext"]["timeEpoch"]
 
 
-class APIGatewayProxyEventV2(dict):
+class APIGatewayProxyEventV2(BaseProxyEvent):
     """AWS Lambda proxy V2 event
 
     Notes:
@@ -418,63 +369,14 @@ def raw_query_string(self) -> str:
     def cookies(self) -> Optional[List[str]]:
         return self.get("cookies")
 
-    @property
-    def headers(self) -> Dict[str, str]:
-        return self["headers"]
-
-    @property
-    def query_string_parameters(self) -> Optional[Dict[str, str]]:
-        return self.get("queryStringParameters")
-
     @property
     def request_context(self) -> RequestContextV2:
         return RequestContextV2(self)
 
-    @property
-    def body(self) -> Optional[str]:
-        return self.get("body")
-
     @property
     def path_parameters(self) -> Optional[Dict[str, str]]:
         return self.get("pathParameters")
 
-    @property
-    def is_base64_encoded(self) -> bool:
-        return self["isBase64Encoded"]
-
     @property
     def stage_variables(self) -> Optional[Dict[str, str]]:
         return self.get("stageVariables")
-
-    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) -> 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
-        Returns
-        -------
-        str, optional
-            Header value
-        """
-        return self.headers.get(name, default_value)

aws_lambda_powertools/utilities/trigger/common.py

@@ -11,3 +11,54 @@ def __getitem__(self, key: str) -> Any:
 
     def get(self, key: str) -> Optional[Any]:
         return self._data.get(key)
+
+
+class BaseProxyEvent(dict):
+    @property
+    def headers(self) -> Dict[str, str]:
+        return self["headers"]
+
+    @property
+    def query_string_parameters(self) -> Optional[Dict[str, str]]:
+        return self.get("queryStringParameters")
+
+    @property
+    def is_base64_encoded(self) -> bool:
+        return self.get("isBase64Encoded")
+
+    @property
+    def body(self) -> Optional[str]:
+        return self.get("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
+        """
+        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) -> 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
+        Returns
+        -------
+        str, optional
+            Header value
+        """
+        return self.headers.get(name, default_value)

tests/events/albEvent.json

@@ -0,0 +1,28 @@
+{
+  "requestContext": {
+    "elb": {
+      "targetGroupArn": "arn:aws:elasticloadbalancing:us-east-2:123456789012:targetgroup/lambda-279XGJDqGZ5rsrHC2Fjr/49e9d65c45c6791a"
+    }
+  },
+  "httpMethod": "GET",
+  "path": "/lambda",
+  "queryStringParameters": {
+    "query": "1234ABCD"
+  },
+  "headers": {
+    "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
+    "accept-encoding": "gzip",
+    "accept-language": "en-US,en;q=0.9",
+    "connection": "keep-alive",
+    "host": "lambda-alb-123578498.us-east-2.elb.amazonaws.com",
+    "upgrade-insecure-requests": "1",
+    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36",
+    "x-amzn-trace-id": "Root=1-5c536348-3d683b8b04734faae651f476",
+    "x-forwarded-for": "72.12.164.125",
+    "x-forwarded-port": "80",
+    "x-forwarded-proto": "http",
+    "x-imforwards": "20"
+  },
+  "body": "Test",
+  "isBase64Encoded": false
+}

tests/functional/test_lambda_trigger_events.py

@@ -4,6 +4,7 @@
 from secrets import compare_digest
 
 from aws_lambda_powertools.utilities.trigger import (
+    ALBEvent,
     APIGatewayProxyEvent,
     APIGatewayProxyEventV2,
     CloudWatchLogsEvent,
@@ -23,6 +24,7 @@
     PreTokenGenerationTriggerEvent,
     UserMigrationTriggerEvent,
 )
+from aws_lambda_powertools.utilities.trigger.common import BaseProxyEvent
 from aws_lambda_powertools.utilities.trigger.dynamo_db_stream_event import (
     AttributeValue,
     DynamoDBRecordEventName,
@@ -531,11 +533,11 @@ def test_api_gateway_proxy_v2_event():
     assert event.stage_variables == event["stageVariables"]
 
 
-def test_api_gateway_proxy_get_query_string_value():
+def test_base_proxy_event_get_query_string_value():
     default_value = "default"
     set_value = "value"
 
-    event = APIGatewayProxyEvent({})
+    event = BaseProxyEvent({})
     value = event.get_query_string_value("test", default_value)
     assert value == default_value
 
@@ -550,49 +552,11 @@ def test_api_gateway_proxy_get_query_string_value():
     assert value is None
 
 
-def test_api_gateway_proxy_v2_get_query_string_value():
+def test_base_proxy_event_get_header_value():
     default_value = "default"
     set_value = "value"
 
-    event = APIGatewayProxyEventV2({})
-    value = event.get_query_string_value("test", default_value)
-    assert value == default_value
-
-    event["queryStringParameters"] = {"test": set_value}
-    value = event.get_query_string_value("test", default_value)
-    assert value == set_value
-
-    value = event.get_query_string_value("unknown", default_value)
-    assert value == default_value
-
-    value = event.get_query_string_value("unknown")
-    assert value is None
-
-
-def test_api_gateway_proxy_get_header_value():
-    default_value = "default"
-    set_value = "value"
-
-    event = APIGatewayProxyEvent({"headers": {}})
-    value = event.get_header_value("test", default_value)
-    assert value == default_value
-
-    event["headers"] = {"test": set_value}
-    value = event.get_header_value("test", default_value)
-    assert value == set_value
-
-    value = event.get_header_value("unknown", default_value)
-    assert value == default_value
-
-    value = event.get_header_value("unknown")
-    assert value is None
-
-
-def test_api_gateway_proxy_v2_get_header_value():
-    default_value = "default"
-    set_value = "value"
-
-    event = APIGatewayProxyEventV2({"headers": {}})
+    event = BaseProxyEvent({"headers": {}})
     value = event.get_header_value("test", default_value)
     assert value == default_value
 
@@ -639,3 +603,16 @@ def test_kinesis_stream_event_json_data():
     data = base64.b64encode(bytes(json.dumps(json_value), "utf-8")).decode("utf-8")
     event = KinesisStreamEvent({"Records": [{"kinesis": {"data": data}}]})
     assert next(event.records).kinesis.data_as_json() == json_value
+
+
+def test_alb_event():
+    event = ALBEvent(load_event("albEvent.json"))
+    assert event.request_context.elb_target_group_arn == event["requestContext"]["elb"]["targetGroupArn"]
+    assert event.http_method == event["httpMethod"]
+    assert event.path == event["path"]
+    assert event.query_string_parameters == event["queryStringParameters"]
+    assert event.headers == event["headers"]
+    assert event.multi_value_query_string_parameters == event.get("multiValueQueryStringParameters")
+    assert event.multi_value_headers == event.get("multiValueHeaders")
+    assert event.body == event["body"]
+    assert event.is_base64_encoded == event["isBase64Encoded"]