chore: refactored multi query code

Author: rubenfonseca

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -369,7 +369,7 @@ def _get_embed_body(
 
 
 def _normalize_multi_query_string_with_param(
-    query_string: Optional[Dict[str, List[str]]],
+    query_string: Dict[str, List[str]],
     params: Sequence[ModelField],
 ) -> Dict[str, Any]:
     """
@@ -386,18 +386,15 @@ def _normalize_multi_query_string_with_param(
     -------
     A dictionary containing the processed multi_query_string_parameters.
     """
-    if not query_string:
-        return {}
-    else:
-        resolved_query_string: Dict[str, Any] = query_string
-        for param in filter(is_scalar_field, params):
-            try:
-                # if the target parameter is a scalar, we keep the first value of the query string
-                # regardless if there are more in the payload
-                resolved_query_string[param.alias] = query_string[param.alias][0]
-            except KeyError:
-                pass
-        return resolved_query_string
+    resolved_query_string: Dict[str, Any] = query_string
+    for param in filter(is_scalar_field, params):
+        try:
+            # if the target parameter is a scalar, we keep the first value of the query string
+            # regardless if there are more in the payload
+            resolved_query_string[param.alias] = query_string[param.alias][0]
+        except KeyError:
+            pass
+    return resolved_query_string
 
 
 def _normalize_multi_header_values_with_param(headers: Optional[Dict[str, str]], params: Sequence[ModelField]):

aws_lambda_powertools/utilities/data_classes/alb_event.py

@@ -36,15 +36,11 @@ def multi_value_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
         return self.get("multiValueQueryStringParameters")
 
     @property
-    def resolved_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
+    def resolved_query_string_parameters(self) -> Dict[str, List[str]]:
         if self.multi_value_query_string_parameters:
             return self.multi_value_query_string_parameters
 
-        if self.query_string_parameters:
-            query_string = {key: value.split(",") for key, value in self.query_string_parameters.items()}
-            return query_string
-
-        return None
+        return super().resolved_query_string_parameters
 
     @property
     def resolved_headers_field(self) -> Optional[Dict[str, Any]]:

aws_lambda_powertools/utilities/data_classes/api_gateway_proxy_event.py

@@ -119,15 +119,11 @@ def multi_value_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
         return self.get("multiValueQueryStringParameters")
 
     @property
-    def resolved_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
+    def resolved_query_string_parameters(self) -> Dict[str, List[str]]:
         if self.multi_value_query_string_parameters:
             return self.multi_value_query_string_parameters
 
-        if self.query_string_parameters:
-            query_string = {key: value.split(",") for key, value in self.query_string_parameters.items()}
-            return query_string
-
-        return None
+        return super().resolved_query_string_parameters
 
     @property
     def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
@@ -322,14 +318,6 @@ def http_method(self) -> str:
     def header_serializer(self):
         return HttpApiHeadersSerializer()
 
-    @property
-    def resolved_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
-        if self.query_string_parameters is not None:
-            query_string = {key: value.split(",") for key, value in self.query_string_parameters.items()}
-            return query_string
-
-        return None
-
     @property
     def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
         if self.headers is not None:

aws_lambda_powertools/utilities/data_classes/bedrock_agent_event.py

@@ -109,14 +109,6 @@ def query_string_parameters(self) -> Optional[Dict[str, str]]:
         # together with the other parameters. So we just return all parameters here.
         return {x["name"]: x["value"] for x in self["parameters"]} if self.get("parameters") else None
 
-    @property
-    def resolved_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
-        if self.query_string_parameters is not None:
-            query_string = {key: value.split(",") for key, value in self.query_string_parameters.items()}
-            return query_string
-
-        return None
-
     @property
     def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
         return {}

aws_lambda_powertools/utilities/data_classes/common.py

@@ -104,7 +104,7 @@ def query_string_parameters(self) -> Optional[Dict[str, str]]:
         return self.get("queryStringParameters")
 
     @property
-    def resolved_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
+    def resolved_query_string_parameters(self) -> Dict[str, List[str]]:
         """
         This property determines the appropriate query string parameter to be used
         as a trusted source for validating OpenAPI.
@@ -116,7 +116,7 @@ def resolved_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
             query_string = {key: value.split(",") for key, value in self.query_string_parameters.items()}
             return query_string
 
-        return None
+        return {}
 
     @property
     def resolved_headers_field(self) -> Optional[Dict[str, Any]]:

aws_lambda_powertools/utilities/data_classes/vpc_lattice.py

@@ -1,5 +1,5 @@
 from functools import cached_property
-from typing import Any, Dict, List, Optional, overload
+from typing import Any, Dict, Optional, overload
 
 from aws_lambda_powertools.shared.headers_serializer import (
     BaseHeadersSerializer,
@@ -138,13 +138,6 @@ def query_string_parameters(self) -> Dict[str, str]:
         """The request query string parameters."""
         return self["query_string_parameters"]
 
-    @property
-    def resolved_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
-        if self.query_string_parameters is not None:
-            query_string = {key: value.split(",") for key, value in self.query_string_parameters.items()}
-            return query_string
-        return None
-
     @property
     def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
         if self.headers is not None:
@@ -256,23 +249,21 @@ def path(self) -> str:
 
     @property
     def request_context(self) -> vpcLatticeEventV2RequestContext:
-        """he VPC Lattice v2 Event request context."""
+        """The VPC Lattice v2 Event request context."""
         return vpcLatticeEventV2RequestContext(self["requestContext"])
 
     @property
     def query_string_parameters(self) -> Optional[Dict[str, str]]:
-        """The request query string parameters."""
-        return self.get("queryStringParameters")
+        """The request query string parameters.
 
-    @property
-    def resolved_query_string_parameters(self) -> Optional[Dict[str, List[str]]]:
-        if self.query_string_parameters is not None:
-            query_string = {
-                key: value.split(",") if not isinstance(value, list) else value
-                for key, value in self.query_string_parameters.items()
-            }
-            return query_string
-        return None
+        For VPC Lattice V2, the queryStringParameters will contain a Dict[str, List[str]]
+        so to keep compatibility with existing utilities, we merge all the values with a comma.
+        """
+        params = self.get("queryStringParameters")
+        if params:
+            return {key: ",".join(value) for key, value in params.items()}
+        else:
+            return None
 
     @property
     def resolved_headers_field(self) -> Optional[Dict[str, str]]: