chore: refactored multi query code

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]]: