support validate_by_name for pydantic BaseModels

Author: tonnico

aws_lambda_powertools/event_handler/api_gateway.py

@@ -889,13 +889,18 @@ def _get_basic_type_schema(param_type: type) -> dict[str, str]:
         """
         Get basic OpenAPI schema for simple types
         """
-        if isinstance(int, param_type):
-            return {"type": "integer"}
-        elif isinstance(float, param_type):
-            return {"type": "number"}
-        elif isinstance(bool, param_type):
-            return {"type": "boolean"}
-        else:
+        try:
+            # Check bool before int, since bool is a subclass of int in Python
+            if issubclass(param_type, bool):
+                return {"type": "boolean"}
+            elif issubclass(param_type, int):
+                return {"type": "integer"}
+            elif issubclass(param_type, float):
+                return {"type": "number"}
+            else:
+                return {"type": "string"}
+        except TypeError:
+            # param_type may not be a type (e.g., typing.Optional[int]), fallback to string
             return {"type": "string"}
 
     @staticmethod

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, ValidationError
+from pydantic import BaseModel
 
 from aws_lambda_powertools.event_handler.middlewares import BaseMiddlewareHandler
 from aws_lambda_powertools.event_handler.openapi.compat import (
@@ -69,8 +69,8 @@ def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) ->
             route.dependant.query_params,
         )
 
-        # Process query values (with Pydantic model support)
-        query_values, query_errors = _request_params_to_args_with_pydantic_support(
+        # Process query values
+        query_values, query_errors = _request_params_to_args(
             route.dependant.query_params,
             query_string,
         )
@@ -81,8 +81,8 @@ def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) ->
             route.dependant.header_params,
         )
 
-        # Process header values (with Pydantic model support)
-        header_values, header_errors = _request_params_to_args_with_pydantic_support(
+        # Process header values
+        header_values, header_errors = _request_params_to_args(
             route.dependant.header_params,
             headers,
         )
@@ -311,7 +311,7 @@ def _prepare_response_content(
         return res  # pragma: no cover
 
 
-def _request_params_to_args_with_pydantic_support(
+def _request_params_to_args(
     required_params: Sequence[ModelField],
     received_params: Mapping[str, Any],
 ) -> tuple[dict[str, Any], list[Any]]:
@@ -330,71 +330,24 @@ 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 - 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
-
+            pass
+        elif isinstance(field_info, Param):
+            pass
         else:
-            # Regular parameter processing (existing logic)
-            if not isinstance(field_info, Param):
-                raise AssertionError(f"Expected Param field_info, got {field_info}")
-
-            value = received_params.get(field.alias)
-            loc = (field_info.in_.value, field.alias)
-
-            if value is None:
-                if field.required:
-                    errors.append(get_missing_field_error(loc=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)
-    return values, errors
-
-
-def _request_params_to_args(
-    required_params: Sequence[ModelField],
-    received_params: Mapping[str, Any],
-) -> tuple[dict[str, Any], list[Any]]:
-    """
-    Convert the request params to a dictionary of values using validation, and returns a list of errors.
-    """
-    values = {}
-    errors = []
-
-    for field in required_params:
-        field_info = field.field_info
-
-        # To ensure early failure, we check if it's not an instance of Param.
-        if not isinstance(field_info, Param):
             raise AssertionError(f"Expected Param field_info, got {field_info}")
 
         value = received_params.get(field.alias)
-
         loc = (field_info.in_.value, field.alias)
 
-        # If we don't have a value, see if it's required or has a default
         if value is None:
             if field.required:
                 errors.append(get_missing_field_error(loc=loc))
             else:
                 values[field.name] = deepcopy(field.default)
             continue
 
-        # Finally, validate the value
+        # Use _validate_field like _request_body_to_args does
         values[field.name] = _validate_field(field=field, value=value, loc=loc, existing_errors=errors)
-
     return values, errors
 
 
@@ -529,7 +482,13 @@ def _normalize_multi_query_string_with_param(
                     try:
                         model_data[field_alias] = query_string[field_alias][0]
                     except KeyError:
-                        pass
+                        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

tests/functional/event_handler/_pydantic/test_openapi_params.py

@@ -80,7 +80,7 @@ def test_openapi_pydantic_header_params():
     class HeaderParams(BaseModel):
         authorization: str = Field(description="Authorization token")
         user_agent: str = Field(default="PowerTools/1.0", description="User agent")
-        accept_language: Optional[str] = Field(default=None, alias="accept-language", description="Language preference")
+        language: Optional[str] = Field(default=None, alias="accept-language", description="Language preference")
 
     @app.get("/protected")
     def protected_handler(headers: Annotated[HeaderParams, Header()]):
@@ -101,7 +101,7 @@ def protected_handler(headers: Annotated[HeaderParams, Header()]):
     # Check individual parameters
     param_names = [param.name for param in get_operation.parameters]
     assert "authorization" in param_names
-    assert "user_agent" in param_names
+    assert "user-agent" in param_names  # headers are always spinal-case
     assert "accept-language" in param_names  # Should use alias
 
     # Check parameter details

tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py

@@ -3,7 +3,7 @@
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import PurePath
-from typing import List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple
 
 import pytest
 from pydantic import BaseModel, Field
@@ -2265,7 +2265,7 @@ def test_validate_pydantic_query_params_with_config_dict_and_validators(gw_event
     """Test that Pydantic models with ConfigDict, aliases, and validators work correctly"""
     from typing import Any
 
-    from pydantic import UUID4, AfterValidator, Base64UrlStr, ConfigDict, StringConstraints, alias_generators
+    from pydantic import AfterValidator, Base64UrlStr, ConfigDict, StringConstraints, alias_generators
 
     del gw_event["multiValueHeaders"]
     del gw_event["multiValueQueryStringParameters"]
@@ -2284,7 +2284,7 @@ class QuerySimple(BaseModel):
         search_id: str
 
     @app.get("/query-model-simple")
-    def query_model(params: Annotated[QuerySimple, Query()]) -> dict[str, Any]:
+    def query_model(params: Annotated[QuerySimple, Query()]) -> Dict[str, Any]:
         return {
             "fullName": params.full_name,
             "nextToken": params.next_token,
@@ -2299,11 +2299,12 @@ class QueryAdvanced(BaseModel):
         model_config = ConfigDict(
             alias_generator=alias_generators.to_camel,
             validate_by_alias=True,
+            validate_by_name=True,
             serialize_by_alias=True,
         )
 
     @app.get("/query-model-advanced")
-    def query_model_advanced(params: Annotated[QueryAdvanced, Query()]) -> dict[str, Any]:
+    def query_model_advanced(params: Annotated[QueryAdvanced, Query()]) -> Dict[str, Any]:
         return params.model_dump()
 
     # Test QuerySimple with validators
@@ -2353,11 +2354,11 @@ def query_model_advanced(params: Annotated[QueryAdvanced, Query()]) -> dict[str,
     assert body["nextToken"] == "dGVzdA=="
     assert body["id"] == "search-456"
 
-    # Test QueryAdvanced with snake_case field names (should also work due to populate_by_name behavior)
+    # Test QueryAdvanced with snake_case field names due to validate_by_name=True
     gw_event["queryStringParameters"] = {
         "full_name": "Snake Case Test",  # snake_case field name
         "next_token": "dGVzdA==",  # snake_case field name
-        "id": "search-789",  # explicit alias
+        "search_id": "search-789",  # snake_case field name
     }
 
     gw_event["path"] = "/query-model-advanced"
@@ -2366,7 +2367,7 @@ def query_model_advanced(params: Annotated[QueryAdvanced, Query()]) -> dict[str,
 
     body = json.loads(result["body"])
     assert body["fullName"] == "Snake Case Test"
-    assert body["nextToken"] == "token789"
+    assert body["nextToken"] == "dGVzdA=="
     assert body["id"] == "search-789"
 
     # Test QueryAdvanced validation error (full_name too short)