diff --git a/aws_lambda_powertools/event_handler/openapi/schema_generator.py b/aws_lambda_powertools/event_handler/openapi/schema_generator.py index 6334a3b53cc..096780996f9 100644 --- a/aws_lambda_powertools/event_handler/openapi/schema_generator.py +++ b/aws_lambda_powertools/event_handler/openapi/schema_generator.py @@ -174,9 +174,15 @@ def _build_operation_parameters( continue if _is_pydantic_model_param(field_info): - parameters.extend(_expand_pydantic_model_parameters(field_info)) + generated_parameters = _expand_pydantic_model_parameters(field_info) else: - parameters.append(_create_regular_parameter(param, model_name_map, field_mapping)) + generated_parameters = [_create_regular_parameter(param, model_name_map, field_mapping)] + + parameters.extend( + parameter + for parameter in generated_parameters + if not (parameter["in"] == "header" and parameter["name"].lower() == "content-type") + ) return parameters diff --git a/tests/functional/event_handler/_pydantic/test_openapi_params.py b/tests/functional/event_handler/_pydantic/test_openapi_params.py index 18087a228d1..40722bb838b 100644 --- a/tests/functional/event_handler/_pydantic/test_openapi_params.py +++ b/tests/functional/event_handler/_pydantic/test_openapi_params.py @@ -1,7 +1,7 @@ import json from dataclasses import dataclass from datetime import datetime -from typing import List, Optional, Tuple +from typing import List, Literal, Optional, Tuple import pytest from pydantic import BaseModel, Field @@ -915,6 +915,24 @@ def mixed_body_endpoint(user_data: Annotated[UserData, Body(media_type="applicat assert "application/json" in request_body.content +def test_openapi_excludes_content_type_header_parameter(): + """Content-Type is described by requestBody content, not an OpenAPI header parameter.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.patch("/json-patch") + def json_patch( + operations: Annotated[list[dict], Body(media_type="application/json-patch+json")], + content_type: Annotated[Literal["application/json-patch+json"], Header(alias="Content-Type")], + ): + return {"status": "updated"} + + schema = app.get_openapi_schema() + patch_op = schema.paths["/json-patch"].patch + + assert "application/json-patch+json" in patch_op.requestBody.content + assert all(parameter.name.lower() != "content-type" for parameter in patch_op.parameters or []) + + def test_openapi_form_parameter_edge_cases(): """Test Form parameters with various edge cases.""" @@ -986,7 +1004,7 @@ def get_items(params: Annotated[QueryParams, Query()]): def test_openapi_pydantic_header_with_alias(): - """Test that Pydantic field aliases work correctly in Header parameters""" + """Test that Pydantic header aliases are emitted, except Content-Type.""" app = APIGatewayRestResolver() class HeaderParams(BaseModel): @@ -1003,7 +1021,7 @@ def test_handler(headers: Annotated[HeaderParams, Header()]): # Check that aliases are used as parameter names param_names = [param.name for param in get_operation.parameters] - assert "content-type" in param_names + assert "content-type" not in param_names assert "user-agent" in param_names assert "content_type" not in param_names # Original field name should not be used assert "user_agent" not in param_names diff --git a/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py b/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py index 16639665c8c..aee94e80b14 100644 --- a/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py +++ b/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py @@ -59,6 +59,27 @@ def handler(user_id: int): assert any(text in result["body"] for text in ["type_error.integer", "int_parsing"]) +def test_content_type_header_validation_remains_enabled(gw_event): + """Content-Type header validation remains available when it is omitted from the OpenAPI schema.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.patch("/json-patch") + def json_patch( + operations: Annotated[list[dict], Body(media_type="application/json-patch+json")], + content_type: Annotated[Literal["application/json-patch+json"], Header(alias="Content-Type")], + ): + return {"status": "updated"} + + gw_event["httpMethod"] = "PATCH" + gw_event["path"] = "/json-patch" + gw_event["headers"]["Content-Type"] = "application/json" + gw_event["body"] = '[{"op": "replace"}]' + + result = app(gw_event, {}) + + assert result["statusCode"] == 422 + + def test_validate_pydantic_query_params(gw_event): """Test that Pydantic models in Query parameters are validated correctly"""