Improve error message and fix header serialization

Author: leandrodamascena

aws_lambda_powertools/event_handler/api_gateway.py

@@ -837,8 +837,14 @@ def _openapi_operation_parameters(
 
                 for field_name, field_def in model_class.model_fields.items():
                     # 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": field_def.alias or field_name,
+                        "name": param_name,
                         "in": field_info.in_.value,
                         "required": field_def.is_required()
                         if hasattr(field_def, "is_required")

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -7,7 +7,7 @@
 from typing import TYPE_CHECKING, Any, Callable, Mapping, MutableMapping, Sequence
 from urllib.parse import parse_qs
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 
 from aws_lambda_powertools.event_handler.middlewares import BaseMiddlewareHandler
 from aws_lambda_powertools.event_handler.openapi.compat import (
@@ -338,6 +338,11 @@ def _request_params_to_args_with_pydantic_support(
             # Extract individual fields from the request
             for model_field_name, model_field_def in model_class.model_fields.items():
                 field_alias = model_field_def.alias or model_field_name
+
+                # Convert snake_case to kebab-case for headers (HTTP convention)
+                if isinstance(field_info, Header):
+                    field_alias = field_alias.replace("_", "-")
+
                 field_value = received_params.get(field_alias)
 
                 if field_value is not None:
@@ -358,8 +363,21 @@ def _request_params_to_args_with_pydantic_support(
                 try:
                     model_instance = model_class(**model_data)
                     values[field.name] = model_instance
+                except ValidationError as e:
+                    # Extract detailed validation errors from Pydantic
+                    for error in e.errors():
+                        # Update the location to include the parameter source (query/header) and field path
+                        error_loc = [field_info.in_.value] + list(error["loc"])
+                        errors.append(
+                            {
+                                "type": error["type"],
+                                "loc": error_loc,
+                                "msg": error["msg"],
+                                "input": error.get("input"),
+                            },
+                        )
                 except Exception as e:
-                    # Validation error
+                    # Fallback for non-Pydantic validation errors
                     loc = (field_info.in_.value, field.alias)
                     errors.append(
                         {
@@ -593,6 +611,10 @@ def _normalize_multi_header_values_with_param(headers: MutableMapping[str, Any],
                     # Normalize individual fields of 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)
+                        field_alias = field_alias.replace("_", "-")
+
                         try:
                             if len(headers[field_alias]) == 1:
                                 headers[field_alias] = headers[field_alias][0]

tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py

@@ -94,6 +94,47 @@ def search_handler(params: Annotated[QueryParams, Query()]):
     assert any("limit" in str(error) for error in body["detail"])
 
 
+def test_validate_pydantic_query_params_detailed_errors(gw_event):
+    """Test that Pydantic validation errors include detailed field-level information"""
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    class QueryParams(BaseModel):
+        full_name: str = Field(..., min_length=5, description="Full name with minimum 5 characters")
+        age: int = Field(..., ge=18, le=100, description="Age between 18 and 100")
+
+    @app.get("/query-model")
+    def query_model(params: Annotated[QueryParams, Query()]):
+        return {"full_name": params.full_name, "age": params.age}
+
+    # Test validation error with detailed field information
+    gw_event["path"] = "/query-model"
+    gw_event["queryStringParameters"] = {"full_name": "Jo", "age": "15"}  # Both invalid
+
+    result = app(gw_event, {})
+    assert result["statusCode"] == 422
+
+    body = json.loads(result["body"])
+    assert "detail" in body
+
+    # Check that we get detailed field-level errors
+    errors = body["detail"]
+
+    # Should have errors for both fields
+    full_name_error = next((e for e in errors if "full_name" in e["loc"]), None)
+    age_error = next((e for e in errors if "age" in e["loc"]), None)
+
+    assert full_name_error is not None, "Should have error for full_name field"
+    assert age_error is not None, "Should have error for age field"
+
+    # Check error details for full_name
+    assert full_name_error["loc"] == ["query", "full_name"]
+    assert full_name_error["type"] == "string_too_short"
+
+    # Check error details for age
+    assert age_error["loc"] == ["query", "age"]
+    assert age_error["type"] == "greater_than_equal"
+
+
 def test_validate_pydantic_header_params(gw_event):
     """Test that Pydantic models in Header parameters are validated correctly"""
     app = APIGatewayRestResolver(enable_validation=True)
@@ -141,6 +182,49 @@ def protected_handler(headers: Annotated[HeaderParams, Header()]):
     assert any("authorization" in str(error) for error in body["detail"])
 
 
+def test_validate_pydantic_header_snake_case_to_kebab_case_schema(gw_event):
+    """Test that snake_case header fields are converted to kebab-case in OpenAPI schema and validation"""
+    app = APIGatewayRestResolver(enable_validation=True)
+    app.enable_swagger()
+
+    class HeaderParams(BaseModel):
+        correlation_id: str = Field(description="Correlation ID header")
+        user_agent: str = Field(default="PowerTools/1.0", description="User agent header")
+
+    @app.get("/kebab-headers")
+    def kebab_handler(headers: Annotated[HeaderParams, Header()]):
+        return {
+            "correlation_id": headers.correlation_id,
+            "user_agent": headers.user_agent,
+        }
+
+    # Test that OpenAPI schema uses kebab-case for headers
+    openapi_schema = app.get_openapi_schema()
+    operation = openapi_schema["paths"]["/kebab-headers"]["get"]
+    parameters = operation["parameters"]
+
+    # Find the correlation_id parameter
+    correlation_param = next((p for p in parameters if p["name"] == "correlation-id"), None)
+    assert correlation_param is not None, "Should have correlation-id parameter in kebab-case"
+    assert correlation_param["in"] == "header"
+
+    # Find the user_agent parameter
+    user_agent_param = next((p for p in parameters if p["name"] == "user-agent"), None)
+    assert user_agent_param is not None, "Should have user-agent parameter in kebab-case"
+    assert user_agent_param["in"] == "header"
+
+    # Test validation with kebab-case headers
+    gw_event["path"] = "/kebab-headers"
+    gw_event["headers"] = {"correlation-id": "test-123", "user-agent": "TestClient/1.0"}
+
+    result = app(gw_event, {})
+    assert result["statusCode"] == 200
+
+    body = json.loads(result["body"])
+    assert body["correlation_id"] == "test-123"
+    assert body["user_agent"] == "TestClient/1.0"
+
+
 def test_validate_pydantic_mixed_params(gw_event):
     """Test that mixed Pydantic models (Query + Header) are validated correctly"""
     app = APIGatewayRestResolver(enable_validation=True)