support multi value headers and queries

Author: tonnico

aws_lambda_powertools/event_handler/api_gateway.py

@@ -839,10 +839,6 @@ def _openapi_operation_parameters(
                     # Create individual parameter for each model field
                     param_name = field_def.alias or field_name
 
-                    # Convert snake_case to kebab-case for headers (HTTP convention)
-                    if isinstance(field_info, Header):
-                        param_name = param_name.replace("_", "-")
-
                     individual_param = {
                         "name": param_name,
                         "in": field_info.in_.value,

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -15,6 +15,7 @@
     _normalize_errors,
     _regenerate_error_with_loc,
     get_missing_field_error,
+    lenient_issubclass,
 )
 from aws_lambda_powertools.event_handler.openapi.dependant import is_scalar_field
 from aws_lambda_powertools.event_handler.openapi.encoders import jsonable_encoder
@@ -64,7 +65,7 @@ def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) ->
         )
 
         # Normalize query values before validate this
-        query_string = _normalize_multi_query_string_with_param(
+        query_string = _normalize_multi_params(
             app.current_event.resolved_query_string_parameters,
             route.dependant.query_params,
         )
@@ -76,7 +77,7 @@ def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) ->
         )
 
         # Normalize header values before validate this
-        headers = _normalize_multi_header_values_with_param(
+        headers = _normalize_multi_params(
             app.current_event.resolved_headers_field,
             route.dependant.header_params,
         )
@@ -439,116 +440,43 @@ def _get_embed_body(
     return received_body, field_alias_omitted
 
 
-def _normalize_multi_query_string_with_param(
-    query_string: dict[str, list[str]],
+def _normalize_multi_params(
+    input_dict: MutableMapping[str, Any],
     params: Sequence[ModelField],
-) -> dict[str, Any]:
+) -> MutableMapping[str, Any]:
     """
-    Extract and normalize resolved_query_string_parameters with Pydantic model support
-
-    Parameters
-    ----------
-    query_string: dict
-        A dictionary containing the initial query string parameters.
-    params: Sequence[ModelField]
-        A sequence of ModelField objects representing parameters.
-
-    Returns
-    -------
-    A dictionary containing the processed multi_query_string_parameters.
+    Generic normalization for query string or header parameters with Pydantic model support.
+    No key transformation is performed.
     """
-    resolved_query_string: dict[str, Any] = query_string
-
     for param in params:
-        # Handle scalar fields (existing logic)
         if is_scalar_field(param):
             try:
-                resolved_query_string[param.alias] = query_string[param.alias][0]
+                val = input_dict[param.alias]
+                if isinstance(val, list) and len(val) == 1:
+                    input_dict[param.alias] = val[0]
+                elif isinstance(val, list):
+                    pass  # leave as list for multi-value
+                # If it's a string, leave as is
             except KeyError:
                 pass
-        # Handle Pydantic models
-        elif isinstance(param.field_info, Query) and hasattr(param.field_info, "annotation"):
-            from pydantic import BaseModel
-
-            from aws_lambda_powertools.event_handler.openapi.compat import lenient_issubclass
-
-            if lenient_issubclass(param.field_info.annotation, BaseModel):
-                model_class = param.field_info.annotation
-                model_data = {}
-
-                # Collect all fields for the Pydantic model
-                for field_name, field_def in model_class.model_fields.items():
-                    field_alias = field_def.alias or field_name
-                    try:
-                        model_data[field_alias] = query_string[field_alias][0]
-                    except KeyError:
-                        if model_class.model_config.get("validate_by_name") or model_class.model_config.get(
-                            "populate_by_name",
-                        ):
-                            try:
-                                model_data[field_alias] = query_string[field_name][0]
-                            except KeyError:
-                                pass
-
-                # Store the collected data under the param alias
-                resolved_query_string[param.alias] = model_data
-
-    return resolved_query_string
-
-
-def _normalize_multi_header_values_with_param(headers: MutableMapping[str, Any], params: Sequence[ModelField]):
-    """
-    Extract and normalize resolved_headers_field with Pydantic model support
-
-    Parameters
-    ----------
-    headers: MutableMapping[str, Any]
-        A dictionary containing the initial header parameters.
-    params: Sequence[ModelField]
-        A sequence of ModelField objects representing parameters.
-
-    Returns
-    -------
-    A dictionary containing the processed headers.
-    """
-    if headers:
-        for param in params:
-            # Handle scalar fields (existing logic)
-            if is_scalar_field(param):
-                try:
-                    if len(headers[param.alias]) == 1:
-                        headers[param.alias] = headers[param.alias][0]
-                except KeyError:
-                    pass
-            # Handle Pydantic models
-            elif isinstance(param.field_info, Header) and hasattr(param.field_info, "annotation"):
-                from pydantic import BaseModel
-
-                from aws_lambda_powertools.event_handler.openapi.compat import lenient_issubclass
-
-                if lenient_issubclass(param.field_info.annotation, BaseModel):
-                    model_class = param.field_info.annotation
-                    model_data = {}
-
-                    # Collect all fields for the Pydantic model
-                    for field_name, field_def in model_class.model_fields.items():
-                        field_alias = field_def.alias or field_name
-
-                        # Convert snake_case to kebab-case for headers (HTTP convention)
-                        header_key = field_alias.replace("_", "-")
-
-                        try:
-                            header_value = headers[header_key]
-                            if isinstance(header_value, list):
-                                if len(header_value) == 1:
-                                    model_data[field_alias] = header_value[0]
-                                else:
-                                    model_data[field_alias] = header_value
-                            else:
-                                model_data[field_alias] = header_value
-                        except KeyError:
-                            pass
-
-                    # Store the collected data under the param alias
-                    headers[param.alias] = model_data
-    return headers
+        elif lenient_issubclass(param.field_info.annotation, BaseModel):
+            model_class = param.field_info.annotation
+            model_data = {}
+            from typing import get_origin
+
+            for field_name, field_def in model_class.model_fields.items():
+                field_alias = field_def.alias or field_name
+                value = input_dict.get(field_alias)
+                if value is None and (
+                    model_class.model_config.get("validate_by_name") or model_class.model_config.get("populate_by_name")
+                ):
+                    value = input_dict.get(field_name)
+                if value is not None:
+                    if get_origin(field_def.annotation) is list:
+                        model_data[field_alias] = value
+                    elif isinstance(value, list):
+                        model_data[field_alias] = value[0]
+                    else:
+                        model_data[field_alias] = value
+            input_dict[param.alias] = model_data
+    return input_dict

aws_lambda_powertools/event_handler/openapi/params.py

@@ -4,7 +4,7 @@
 from enum import Enum
 from typing import TYPE_CHECKING, Any, Literal
 
-from pydantic import BaseConfig
+from pydantic import BaseConfig, BaseModel, create_model
 from pydantic.fields import FieldInfo
 from typing_extensions import Annotated, get_args, get_origin
 
@@ -17,6 +17,7 @@
     copy_field_info,
     field_annotation_is_scalar,
     get_annotation_from_field_info,
+    lenient_issubclass,
 )
 
 if TYPE_CHECKING:
@@ -1094,6 +1095,42 @@ def create_response_field(
     return ModelField(**kwargs)  # type: ignore[arg-type]
 
 
+def _apply_header_underscore_conversion(
+    field_info: FieldInfo,
+    type_annotation: Any,
+    param_name: str,
+) -> tuple[FieldInfo, Any]:
+    """
+    Apply underscore-to-dash conversion for Header parameters.
+
+    For BaseModel: Creates new model with underscore-to-dash alias generator.
+    Note: If the BaseModel already has an alias generator, it will be replaced
+    with dash-case conversion since HTTP headers should use dash-case.
+    For all Header fields: Sets the parameter alias if convert_underscores is True
+    """
+    if not isinstance(field_info, Header) or not field_info.convert_underscores:
+        return field_info, type_annotation
+
+    # Always set the parameter alias for Header fields (if not already set)
+    if not field_info.alias:
+        field_info.alias = param_name.replace("_", "-")
+
+    # Handle BaseModel case - create new model with dash-case alias generator
+    if lenient_issubclass(type_annotation, BaseModel):
+        # For HTTP headers, we should use dash-case regardless of existing alias generator
+        # This ensures consistent header naming conventions
+        header_aliased_model = create_model(
+            f"{type_annotation.__name__}WithHeaderAliases",
+            __base__=type_annotation,
+            __config__={"alias_generator": lambda name: name.replace("_", "-")},
+        )
+
+        type_annotation = header_aliased_model
+        field_info.annotation = type_annotation
+
+    return field_info, type_annotation
+
+
 def _create_model_field(
     field_info: FieldInfo | None,
     type_annotation: Any,
@@ -1112,21 +1149,17 @@ def _create_model_field(
     elif isinstance(field_info, Param) and getattr(field_info, "in_", None) is None:
         field_info.in_ = ParamTypes.query
 
+    # Apply header underscore conversion
+    field_info, type_annotation = _apply_header_underscore_conversion(field_info, type_annotation, param_name)
+
     # If the field_info is a Param, we use the `in_` attribute to determine the type annotation
     use_annotation = get_annotation_from_field_info(type_annotation, field_info, param_name)
 
-    # If the field doesn't have a defined alias, we use the param name
-    if not field_info.alias and getattr(field_info, "convert_underscores", None):
-        alias = param_name.replace("_", "-")
-    else:
-        alias = field_info.alias or param_name
-    field_info.alias = alias
-
     return create_response_field(
         name=param_name,
         type_=use_annotation,
         default=field_info.default,
-        alias=alias,
+        alias=field_info.alias,
         required=field_info.default in (Required, Undefined),
         field_info=field_info,
     )