feat: Add a flag to ALBResolver to URL-decode query parameters

Author: chriselion

aws_lambda_powertools/event_handler/api_gateway.py

@@ -3154,6 +3154,7 @@ 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"""
         super().__init__(
@@ -3166,6 +3167,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
@@ -3192,3 +3194,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",
     [

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}