Improving field validation method + tests

Author: leandrodamascena

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -330,63 +330,21 @@ def _request_params_to_args_with_pydantic_support(
         from aws_lambda_powertools.event_handler.openapi.compat import lenient_issubclass
 
         if isinstance(field_info, (Query, Header)) and lenient_issubclass(field_info.annotation, BaseModel):
-            # Handle Pydantic model
-            model_class = field_info.annotation
-            model_data = {}
-            model_errors = []
-
-            # 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:
-                    model_data[model_field_name] = field_value
-                elif (
-                    model_field_def.is_required()
-                    if hasattr(model_field_def, "is_required")
-                    else model_field_def.default is ...
-                ):
-                    # Required field missing
-                    loc = (field_info.in_.value, field_alias)
-                    model_errors.append(get_missing_field_error(loc=loc))
-
-            if model_errors:
-                errors.extend(model_errors)
-            else:
-                # Try to create the Pydantic model
-                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:
-                    # Fallback for non-Pydantic validation errors
-                    loc = (field_info.in_.value, field.alias)
-                    errors.append(
-                        {
-                            "type": "value_error",
-                            "loc": loc,
-                            "msg": str(e),
-                            "input": model_data,
-                        },
-                    )
+            # Handle Pydantic model - use the same approach as _request_body_to_args
+            loc = (field_info.in_.value, field.alias)
+
+            # Get the raw data for the Pydantic model
+            value = received_params.get(field.alias)
+
+            if value is None:
+                if field.required:
+                    errors.append(get_missing_field_error(loc))
+                else:
+                    values[field.name] = deepcopy(field.default)
+                continue
+
+            # Use _validate_field like _request_body_to_args does
+            values[field.name] = _validate_field(field=field, value=value, loc=loc, existing_errors=errors)
         else:
             # Regular parameter processing (existing logic)
             if not isinstance(field_info, Param):
@@ -565,14 +523,19 @@ def _normalize_multi_query_string_with_param(
 
             if lenient_issubclass(param.field_info.annotation, BaseModel):
                 model_class = param.field_info.annotation
-                # Normalize individual fields of the Pydantic model
+                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:
-                        resolved_query_string[field_alias] = query_string[field_alias][0]
+                        model_data[field_alias] = query_string[field_alias][0]
                     except KeyError:
                         pass
 
+                # Store the collected data under the param alias
+                resolved_query_string[param.alias] = model_data
+
     return resolved_query_string
 
 
@@ -608,16 +571,27 @@ def _normalize_multi_header_values_with_param(headers: MutableMapping[str, Any],
 
                 if lenient_issubclass(param.field_info.annotation, BaseModel):
                     model_class = param.field_info.annotation
-                    # Normalize individual fields of the Pydantic model
+                    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)
-                        field_alias = field_alias.replace("_", "-")
+                        header_key = field_alias.replace("_", "-")
 
                         try:
-                            if len(headers[field_alias]) == 1:
-                                headers[field_alias] = headers[field_alias][0]
+                            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