fix(event_handler): unify header validation, scalar list joining, and content-type parsing

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -29,7 +29,6 @@
 from aws_lambda_powertools.event_handler.openapi.params import Param, UploadFile
 from aws_lambda_powertools.event_handler.openapi.types import UnionType
 from aws_lambda_powertools.utilities.data_classes.alb_event import ALBEvent
-from aws_lambda_powertools.utilities.data_classes.common import CaseInsensitiveDict
 from aws_lambda_powertools.utilities.data_classes.vpc_lattice import VPCLatticeEventV2
 
 if TYPE_CHECKING:
@@ -97,7 +96,6 @@ def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) ->
         headers = _normalize_multi_params(
             app.current_event.resolved_headers_field,
             route.dependant.header_params,
-            multi_value_params=CaseInsensitiveDict(app.current_event.get("multiValueHeaders")),
         )
 
         # Process header values
@@ -142,17 +140,18 @@ def _get_body(self, app: EventHandlerInstance) -> dict[str, Any]:
         Get the request body from the event, and parse it according to content type.
         """
         content_type = app.current_event.headers.get("content-type", "").strip()
+        media_type = content_type.split(";")[0].strip().lower()
 
         # Handle JSON content
-        if not content_type or content_type.startswith(APPLICATION_JSON_CONTENT_TYPE):
+        if not media_type or media_type == APPLICATION_JSON_CONTENT_TYPE or media_type.endswith("+json"):
             return self._parse_json_data(app)
 
         # Handle URL-encoded form data
-        elif content_type.startswith(APPLICATION_FORM_CONTENT_TYPE):
+        elif media_type == APPLICATION_FORM_CONTENT_TYPE:
             return self._parse_form_data(app)
 
         # Handle multipart/form-data (file uploads)
-        elif content_type.startswith(MULTIPART_FORM_DATA_CONTENT_TYPE):
+        elif media_type == MULTIPART_FORM_DATA_CONTENT_TYPE:
             return self._parse_multipart_data(app, content_type)
 
         else:
@@ -636,6 +635,8 @@ def _normalize_multi_params(
     for param in params:
         if is_scalar_field(param):
             _process_scalar_param(input_dict, param, multi_value_params)
+        elif _is_or_contains_sequence(param.field_info.annotation):
+            _process_sequence_param(input_dict, param)
         elif lenient_issubclass(param.field_info.annotation, BaseModel):
             _process_model_param(input_dict, param, multi_value_params)
     return input_dict
@@ -668,6 +669,21 @@ def _process_scalar_param(
         pass
 
 
+def _process_sequence_param(
+    input_dict: MutableMapping[str, Any],
+    param: ModelField,
+) -> None:
+    """Restore sequence parameters by splitting comma strings or wrapping single values in a list."""
+    try:
+        value = input_dict[param.alias]
+    except KeyError:
+        return
+
+    # headers arrive as a single string (possibly comma-joined), query strings already as a list
+    if isinstance(value, str):
+        input_dict[param.alias] = value.split(",")
+
+
 def _process_model_param(
     input_dict: MutableMapping[str, Any],
     param: ModelField,
@@ -685,6 +701,8 @@ def _process_model_param(
             if not _is_or_contains_sequence(field_info.annotation):
                 source_name = field_alias if input_dict.get(field_alias) is not None else field_name
                 value = _restore_scalar_parameter(value, source_name, multi_value_params)
+            elif isinstance(value, str):
+                value = value.split(",")
             model_data[field_alias] = _normalize_field_value(value=value, field_info=field_info)
 
     input_dict[param.alias] = model_data

tests/functional/event_handler/_pydantic/test_openapi_comma_parameters.py

@@ -242,7 +242,10 @@ def scalar_handler(search: Annotated[str, location()]):
 
     result = app.resolve(event, {})
 
-    if model:
+    if location is Header:
+        assert result["statusCode"] == 200
+        assert json.loads(result["body"]) == {"search": "first,second"}
+    elif model:
         assert result["statusCode"] == 200
         assert json.loads(result["body"]) == {"search": "first"}
     else:

tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py

@@ -4390,3 +4390,131 @@ def handler() -> Response[None]:
     result = app(gw_event_alb, {})
     assert result["statusCode"] == 204
     assert result["body"] == ""
+
+
+def test_header_sequence_single_value_across_resolvers(gw_event_http):
+    # GIVEN an APIGatewayHttpResolver with validation enabled and a single-value sequence header
+    app = APIGatewayHttpResolver(enable_validation=True)
+    gw_event_http["rawPath"] = gw_event_http["requestContext"]["http"]["path"] = "/hello"
+    gw_event_http["requestContext"]["http"]["method"] = "GET"
+    gw_event_http["headers"] = {"x-country": "US"}
+
+    @app.get("/hello")
+    def hello(x_country: Annotated[list[str], Header(alias="x-country")]):
+        return {"c": x_country}
+
+    result = app(gw_event_http, {})
+    assert result["statusCode"] == 200
+    assert json.loads(result["body"]) == {"c": ["US"]}
+
+
+def test_repeated_scalar_header_joined_across_resolvers(gw_event):
+    # GIVEN an APIGatewayRestResolver with validation enabled and repeated scalar headers
+    app = APIGatewayRestResolver(enable_validation=True)
+    gw_event["path"] = "/h"
+    gw_event["httpMethod"] = "GET"
+    gw_event["headers"] = {"X-Req": "a"}
+    gw_event["multiValueHeaders"] = {"X-Req": ["a", "b"]}
+
+    @app.get("/h")
+    def h(x_req: Annotated[str, Header()]):
+        return {"x": x_req}
+
+    result = app(gw_event, {})
+    assert result["statusCode"] == 200
+    assert json.loads(result["body"]) == {"x": "a,b"}
+
+
+@pytest.mark.parametrize(
+    "content_type",
+    [
+        "Application/JSON",
+        "application/json; charset=utf-8",
+        "Application/JSON; charset=utf-8",
+        "application/vnd.api+json",
+        "application/problem+json",
+        "APPLICATION/PROBLEM+JSON; charset=utf-8",
+    ],
+)
+def test_content_type_case_insensitive_and_json_suffix(gw_event, content_type):
+    # GIVEN an APIGatewayRestResolver with validation enabled and case-insensitive / +json media types
+    class Item(BaseModel):
+        name: str
+
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.post("/items")
+    def create(item: Item):
+        return item.model_dump()
+
+    gw_event["path"] = "/items"
+    gw_event["httpMethod"] = "POST"
+    gw_event["body"] = json.dumps({"name": "powertools"})
+    gw_event["headers"] = {"Content-Type": content_type}
+    gw_event["multiValueHeaders"] = {"Content-Type": [content_type]}
+
+    result = app(gw_event, {})
+    assert result["statusCode"] == 200
+    assert json.loads(result["body"]) == {"name": "powertools"}
+
+
+def test_header_sequence_optional_absent_uses_default(gw_event_http):
+    # GIVEN an APIGatewayHttpResolver with validation enabled and an optional sequence header that is not sent
+    app = APIGatewayHttpResolver(enable_validation=True)
+    gw_event_http["rawPath"] = gw_event_http["requestContext"]["http"]["path"] = "/hello"
+    gw_event_http["requestContext"]["http"]["method"] = "GET"
+    gw_event_http["headers"] = {}
+
+    @app.get("/hello")
+    def hello(x_country: Annotated[list[str] | None, Header(alias="x-country")] = None):
+        return {"c": x_country}
+
+    # WHEN the request has no such header
+    result = app(gw_event_http, {})
+
+    # THEN the default is used instead of failing validation
+    assert result["statusCode"] == 200
+    assert json.loads(result["body"]) == {"c": None}
+
+
+def test_header_model_sequence_field_single_value(gw_event_http):
+    # GIVEN a Pydantic header model with a sequence field and a single header value on HTTP API
+    class HeaderModel(BaseModel):
+        x_country: list[str] = Field(alias="x-country")
+        x_trace: str = Field(alias="x-trace")
+
+    app = APIGatewayHttpResolver(enable_validation=True)
+    gw_event_http["rawPath"] = gw_event_http["requestContext"]["http"]["path"] = "/hello"
+    gw_event_http["requestContext"]["http"]["method"] = "GET"
+    gw_event_http["headers"] = {"x-country": "US", "x-trace": "t-1"}
+
+    @app.get("/hello")
+    def hello(headers: Annotated[HeaderModel, Header()]):
+        return {"c": headers.x_country, "t": headers.x_trace}
+
+    # WHEN the request carries one value for the sequence field
+    result = app(gw_event_http, {})
+
+    # THEN it is accepted as a one-item list and the scalar field is untouched
+    assert result["statusCode"] == 200
+    assert json.loads(result["body"]) == {"c": ["US"], "t": "t-1"}
+
+
+def test_header_model_sequence_field_comma_joined(gw_event_http):
+    # GIVEN a Pydantic header model with a sequence field and a comma-joined header value
+    class HeaderModel(BaseModel):
+        x_country: list[str] = Field(alias="x-country")
+
+    app = APIGatewayHttpResolver(enable_validation=True)
+    gw_event_http["rawPath"] = gw_event_http["requestContext"]["http"]["path"] = "/hello"
+    gw_event_http["requestContext"]["http"]["method"] = "GET"
+    gw_event_http["headers"] = {"x-country": "US,PT"}
+
+    @app.get("/hello")
+    def hello(headers: Annotated[HeaderModel, Header()]):
+        return {"c": headers.x_country}
+
+    result = app(gw_event_http, {})
+
+    assert result["statusCode"] == 200
+    assert json.loads(result["body"]) == {"c": ["US", "PT"]}