feat(trigger): data class and helper functions for lambda trigger events

aws_lambda_powertools/utilities/trigger/__init__.py

@@ -0,0 +1,24 @@
+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
+from .event_bridge_event import EventBridgeEvent
+from .kinesis_stream_event import KinesisStreamEvent
+from .s3_event import S3Event
+from .ses_event import SESEvent
+from .sns_event import SNSEvent
+from .sqs_event import SQSEvent
+
+__all__ = [
+    "APIGatewayProxyEvent",
+    "APIGatewayProxyEventV2",
+    "ALBEvent",
+    "CloudWatchLogsEvent",
+    "DynamoDBStreamEvent",
+    "EventBridgeEvent",
+    "KinesisStreamEvent",
+    "S3Event",
+    "SESEvent",
+    "SNSEvent",
+    "SQSEvent",
+]

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")