feat(event_handlers): add support for Lambda Function URL

Author: rubenfonseca

aws_lambda_powertools/event_handler/__init__.py

@@ -11,6 +11,7 @@
     Response,
 )
 from .appsync import AppSyncResolver
+from .lambda_function_url import LambdaFunctionUrlResolver
 
 __all__ = [
     "AppSyncResolver",
@@ -19,5 +20,6 @@
     "ALBResolver",
     "ApiGatewayResolver",
     "CORSConfig",
+    "LambdaFunctionUrlResolver",
     "Response",
 ]

aws_lambda_powertools/event_handler/api_gateway.py

@@ -17,7 +17,12 @@
 from aws_lambda_powertools.shared import constants
 from aws_lambda_powertools.shared.functions import resolve_truthy_env_var_choice
 from aws_lambda_powertools.shared.json_encoder import Encoder
-from aws_lambda_powertools.utilities.data_classes import ALBEvent, APIGatewayProxyEvent, APIGatewayProxyEventV2
+from aws_lambda_powertools.utilities.data_classes import (
+    ALBEvent,
+    APIGatewayProxyEvent,
+    APIGatewayProxyEventV2,
+    LambdaFunctionUrlEvent,
+)
 from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent
 from aws_lambda_powertools.utilities.typing import LambdaContext
 
@@ -36,6 +41,7 @@ class ProxyEventType(Enum):
     APIGatewayProxyEvent = "APIGatewayProxyEvent"
     APIGatewayProxyEventV2 = "APIGatewayProxyEventV2"
     ALBEvent = "ALBEvent"
+    LambdaFunctionUrlEvent = "LambdaFunctionUrlEvent"
 
 
 class CORSConfig:
@@ -546,6 +552,9 @@ def _to_proxy_event(self, event: Dict) -> BaseProxyEvent:
         if self._proxy_type == ProxyEventType.APIGatewayProxyEventV2:
             logger.debug("Converting event to API Gateway HTTP API contract")
             return APIGatewayProxyEventV2(event)
+        if self._proxy_type == ProxyEventType.LambdaFunctionUrlEvent:
+            logger.debug("Converting event to Lambda Function URL contract")
+            return LambdaFunctionUrlEvent(event)
         logger.debug("Converting event to ALB contract")
         return ALBEvent(event)
 

aws_lambda_powertools/event_handler/lambda_function_url.py

@@ -0,0 +1,18 @@
+from typing import Callable, Dict, List, Optional
+
+from aws_lambda_powertools.event_handler import CORSConfig
+from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver, ProxyEventType
+from aws_lambda_powertools.utilities.data_classes import LambdaFunctionUrlEvent
+
+
+class LambdaFunctionUrlResolver(ApiGatewayResolver):
+    current_event: LambdaFunctionUrlEvent
+
+    def __init__(
+        self,
+        cors: Optional[CORSConfig] = None,
+        debug: Optional[bool] = None,
+        serializer: Optional[Callable[[Dict], str]] = None,
+        strip_prefixes: Optional[List[str]] = None,
+    ):
+        super().__init__(ProxyEventType.LambdaFunctionUrlEvent, cors, debug, serializer, strip_prefixes)

aws_lambda_powertools/logging/correlation_paths.py

@@ -6,4 +6,5 @@
 APPSYNC_RESOLVER = 'request.headers."x-amzn-trace-id"'
 APPLICATION_LOAD_BALANCER = 'headers."x-amzn-trace-id"'
 EVENT_BRIDGE = "id"
+LAMBDA_FUNCTION_URL = "requestContext.requestId"
 S3_OBJECT_LAMBDA = "xAmzRequestId"

aws_lambda_powertools/utilities/data_classes/lambda_function_url_event.py

@@ -0,0 +1,19 @@
+from aws_lambda_powertools.utilities.data_classes.api_gateway_proxy_event import APIGatewayProxyEventV2
+
+
+class LambdaFunctionUrlEvent(APIGatewayProxyEventV2):
+    """AWS Lambda Function URL event
+
+    Notes:
+    -----
+    For now, this seems to follow the exact payload as HTTP APIs Payload Format Version 2.0.
+    Certain keys in this payload format don't make sense for function urls (e.g: `routeKey`).
+    These keys will always be null.
+
+    Documentation:
+    - https://docs.aws.amazon.com/lambda/latest/dg/urls-invocation.html#urls-payloads
+    - https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-integrations-lambda.html
+
+    """
+
+    pass

tests/functional/event_handler/test_lambda_function_url.py

@@ -0,0 +1,40 @@
+from aws_lambda_powertools.event_handler import LambdaFunctionUrlResolver, Response, content_types
+from aws_lambda_powertools.utilities.data_classes import LambdaFunctionUrlEvent
+from tests.functional.utils import load_event
+
+
+def test_lambda_function_url_event():
+    # GIVEN a Lambda Function Url type event
+    app = LambdaFunctionUrlResolver()
+
+    @app.post("/my/path")
+    def foo():
+        assert isinstance(app.current_event, LambdaFunctionUrlEvent)
+        assert app.lambda_context == {}
+        assert app.current_event.request_context.stage is not None
+        return Response(200, content_types.TEXT_HTML, "foo")
+
+    # WHEN calling the event handler
+    result = app(load_event("lambdaFunctionUrlEvent.json"), {})
+
+    # THEN process event correctly
+    # AND set the current_event type as LambdaFunctionUrlEvent
+    assert result["statusCode"] == 200
+    assert result["headers"]["Content-Type"] == content_types.TEXT_HTML
+    assert result["body"] == "foo"
+
+
+def test_lambda_function_url_no_matches():
+    # GIVEN a Lambda Function Url type event
+    app = LambdaFunctionUrlResolver()
+
+    @app.post("/no_match")
+    def foo():
+        raise RuntimeError()
+
+    # WHEN calling the event handler
+    result = app(load_event("lambdaFunctionUrlEvent.json"), {})
+
+    # THEN process event correctly
+    # AND return 404 because the event doesn't match any known route
+    assert result["statusCode"] == 404