Merge branch 'develop' into fix/openapi-response-schema-bleed

Author: leandrodamascena

aws_lambda_powertools/event_handler/api_gateway.py

@@ -3318,8 +3318,34 @@ def __init__(
         enable_validation: bool = False,
         response_validation_error_http_code: HTTPStatus | int | None = None,
         json_body_deserializer: Callable[[str], dict] | None = None,
+        decode_query_parameters: bool = False,
     ):
-        """Amazon Application Load Balancer (ALB) resolver"""
+        """Amazon Application Load Balancer (ALB) resolver
+
+
+        Parameters
+        ----------
+        cors: CORSConfig
+            Optionally configure and enabled CORS. Not each route will need to have to cors=True
+        debug: bool | None
+            Enables debug mode, by default False. Can be also be enabled by "POWERTOOLS_DEV"
+            environment variable
+        serializer: Callable, optional
+            function to serialize `obj` to a JSON formatted `str`, by default json.dumps
+        strip_prefixes: list[str | Pattern], optional
+            optional list of prefixes to be removed from the request path before doing the routing.
+            This is often used with api gateways with multiple custom mappings.
+            Each prefix can be a static string or a compiled regex pattern
+        enable_validation: bool | None
+            Enables validation of the request body against the route schema, by default False.
+        response_validation_error_http_code
+            Sets the returned status code if response is not validated. enable_validation must be True.
+        json_body_deserializer: Callable[[str], dict], optional
+            function to deserialize `str`, `bytes`, `bytearray` containing a JSON document to a Python `dict`,
+            by default json.loads when integrating with EventSource data class
+        decode_query_parameters: bool | None
+            Enables URL-decoding of query parameters (both keys and values), by default False.
+        """
         super().__init__(
             ProxyEventType.ALBEvent,
             cors,
@@ -3330,6 +3356,7 @@ def __init__(
             response_validation_error_http_code,
             json_body_deserializer=json_body_deserializer,
         )
+        self.decode_query_parameters = decode_query_parameters
 
     def _get_base_path(self) -> str:
         # ALB doesn't have a stage variable, so we just return an empty string
@@ -3356,3 +3383,10 @@ def _to_response(self, result: dict | tuple | Response | BedrockResponse) -> Res
             result.body = ""
 
         return super()._to_response(result)
+
+    @override
+    def _to_proxy_event(self, event: dict) -> BaseProxyEvent:
+        proxy_event = super()._to_proxy_event(event)
+        if isinstance(proxy_event, ALBEvent):
+            proxy_event.decode_query_parameters = self.decode_query_parameters
+        return proxy_event

aws_lambda_powertools/utilities/data_classes/alb_event.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Callable
+from urllib.parse import unquote
+
+from typing_extensions import override
 
 from aws_lambda_powertools.shared.headers_serializer import (
     BaseHeadersSerializer,
@@ -30,13 +33,27 @@ class ALBEvent(BaseProxyEvent):
     - https://docs.aws.amazon.com/elasticloadbalancing/latest/application/lambda-functions.html
     """
 
+    @override
+    def __init__(self, data: dict[str, Any], json_deserializer: Callable | None = None):
+        super().__init__(data, json_deserializer)
+        self.decode_query_parameters = False
+
     @property
     def request_context(self) -> ALBEventRequestContext:
         return ALBEventRequestContext(self["requestContext"])
 
     @property
     def resolved_query_string_parameters(self) -> dict[str, list[str]]:
-        return self.multi_value_query_string_parameters or super().resolved_query_string_parameters
+        params = self.multi_value_query_string_parameters or super().resolved_query_string_parameters
+        if not self.decode_query_parameters:
+            return params
+
+        # Decode the parameter keys and values
+        decoded_params = {}
+        for k, vals in params.items():
+            decoded_params[unquote(k)] = [unquote(v) for v in vals]
+
+        return decoded_params
 
     @property
     def multi_value_headers(self) -> dict[str, list[str]]:

tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py

@@ -1,4 +1,5 @@
 import base64
+import datetime
 import json
 from dataclasses import dataclass
 from enum import Enum
@@ -20,6 +21,7 @@
 )
 from aws_lambda_powertools.event_handler.openapi.exceptions import ResponseValidationError
 from aws_lambda_powertools.event_handler.openapi.params import Body, Form, Header, Query
+from tests.functional.utils import load_event
 
 
 def test_validate_scalars(gw_event):
@@ -1070,6 +1072,26 @@ def handler3():
         assert any(text in result["body"] for text in expected_error_text)
 
 
+def test_validation_query_string_with_encoded_datetime_alb_resolver():
+    # GIVEN a ALBResolver with validation enabled,
+    # and an event with a url-encoded datetime
+    # as a query string parameter
+    app = ALBResolver(enable_validation=True, decode_query_parameters=True)
+    raw_event = load_event("albEvent.json")
+    raw_event["path"] = "/users"
+    raw_event["queryStringParameters"] = {"query_dt": "2025-12-20T16%3A56%3A02.032000"}
+
+    # WHEN a handler is defined with various parameters and routes
+    @app.get("/users")
+    def handler(query_dt: datetime.datetime):
+        return None
+
+    # THEN the handler should be invoked with the expected result
+    # AND the status code should match the expected_status_code
+    result = app(raw_event, {})
+    assert result["statusCode"] == 200
+
+
 @pytest.mark.parametrize(
     "handler_func, expected_status_code, expected_error_text",
     [
@@ -2672,3 +2694,60 @@ def query_model_advanced(params: Annotated[QueryAdvanced, Query()]) -> Dict[str,
     full_name_error = next((e for e in errors if "full_name" in e["loc"] or "fullName" in e["loc"]), None)
     assert full_name_error is not None
     assert full_name_error["type"] == "string_too_short"
+
+
+def test_validation_query_string_with_fully_encoded_datetime_alb_resolver():
+    # GIVEN a ALBResolver with validation enabled,
+    # and an event with a fully url-encoded datetime
+    # as a query string parameter
+    app = ALBResolver(enable_validation=True, decode_query_parameters=True)
+    raw_event = load_event("albEvent.json")
+    raw_event["path"] = "/users"
+    # Fully encoded: "2025-12-20T16:56:02.032000" -> "2025-12-20T16%3A56%3A02.032000"
+    # With spaces or special chars: "2025-12-20 16:56:02" -> "2025-12-20%2016%3A56%3A02"
+    raw_event["queryStringParameters"] = {"query_dt": "2025-12-20T16%3A56%3A02.032000"}
+
+    @app.get("/users")
+    def handler(query_dt: datetime.datetime):
+        return {"received": query_dt.isoformat()}
+
+    result = app(raw_event, {})
+    assert result["statusCode"] == 200
+    body = json.loads(result["body"])
+    assert body["received"] == "2025-12-20T16:56:02.032000"
+
+
+def test_validation_query_string_with_encoded_key_and_value_alb_resolver():
+    # GIVEN a ALBResolver with validation enabled,
+    # and an event with url-encoded key AND value
+    app = ALBResolver(enable_validation=True, decode_query_parameters=True)
+    raw_event = load_event("albEvent.json")
+    raw_event["path"] = "/search"
+    # Key: "search query" -> "search%20query"
+    # Value: "hello world" -> "hello%20world"
+    raw_event["queryStringParameters"] = {"search%20query": "hello%20world"}
+
+    @app.get("/search")
+    def handler(search_query: Annotated[str, Query(alias="search query")]):
+        return {"result": search_query}
+
+    result = app(raw_event, {})
+    assert result["statusCode"] == 200
+    body = json.loads(result["body"])
+    assert body["result"] == "hello world"
+
+
+def test_validation_without_decode_query_parameters_alb_resolver():
+    # GIVEN a ALBResolver WITHOUT decode_query_parameters (default behavior)
+    app = ALBResolver(enable_validation=True)
+    raw_event = load_event("albEvent.json")
+    raw_event["path"] = "/users"
+    raw_event["queryStringParameters"] = {"query_dt": "2025-12-20T16%3A56%3A02.032000"}
+
+    @app.get("/users")
+    def handler(query_dt: datetime.datetime):
+        return None
+
+    # THEN validation should fail because the encoded string is not a valid datetime
+    result = app(raw_event, {})
+    assert result["statusCode"] == 422

tests/unit/data_classes/required_dependencies/test_alb_event.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from urllib.parse import quote
+
 from aws_lambda_powertools.utilities.data_classes import ALBEvent
 from tests.functional.utils import load_event
 
@@ -19,3 +21,34 @@ def test_alb_event():
     assert parsed_event.multi_value_headers == (raw_event.get("multiValueHeaders") or {})
     assert parsed_event.body == raw_event["body"]
     assert parsed_event.is_base64_encoded == raw_event["isBase64Encoded"]
+
+
+def test_alb_event_decode_query_parameters():
+    expected_key = "this is a key"
+    expected_value = "single value"
+    raw_event = load_event("albEvent.json")
+    raw_event["queryStringParameters"] = {quote(expected_key): quote(expected_value)}
+    # Without decode_query_parameters, the key and value are not decoded
+    parsed_event = ALBEvent(raw_event)
+    assert parsed_event.resolved_query_string_parameters != {expected_key: [expected_value]}
+    assert parsed_event.resolved_query_string_parameters == {quote(expected_key): [quote(expected_value)]}
+
+    # With decode_query_parameters, the key and value are not decoded
+    parsed_event.decode_query_parameters = True
+    assert parsed_event.resolved_query_string_parameters == {expected_key: [expected_value]}
+
+
+def test_alb_event_decode_multi_value_query_parameters():
+    expected_key = "this is a key"
+    expected_values = ["first value", "second value"]
+
+    raw_event = load_event("albMultiValueQueryStringEvent.json")
+    raw_event["multiValueQueryStringParameters"] = {quote(expected_key): [quote(v) for v in expected_values]}
+    # Without decode_query_parameters, the key and value are not decoded
+    parsed_event = ALBEvent(raw_event)
+    assert parsed_event.resolved_query_string_parameters != {expected_key: expected_values}
+    assert parsed_event.resolved_query_string_parameters == {quote(expected_key): [quote(v) for v in expected_values]}
+
+    # With decode_query_parameters, the key and value are not decoded
+    parsed_event.decode_query_parameters = True
+    assert parsed_event.resolved_query_string_parameters == {expected_key: expected_values}