Urlencoded deserializer schema matching type coercion

Author: p1c2u

openapi_core/deserializing/media_types/deserializers.py

@@ -1,3 +1,4 @@
+from typing import TYPE_CHECKING
 from typing import Any
 from typing import Mapping
 from typing import Optional
@@ -22,6 +23,10 @@
 from openapi_core.schema.protocols import SuportsGetAll
 from openapi_core.schema.protocols import SuportsGetList
 from openapi_core.schema.schemas import get_properties
+from openapi_core.validation.schemas.validators import SchemaValidator
+
+if TYPE_CHECKING:
+    from openapi_core.casting.schemas.casters import SchemaCaster
 
 
 class MediaTypesDeserializer:
@@ -65,13 +70,17 @@ def __init__(
         media_types_deserializer: MediaTypesDeserializer,
         mimetype: str,
         schema: Optional[SchemaPath] = None,
+        schema_validator: Optional[SchemaValidator] = None,
+        schema_caster: Optional["SchemaCaster"] = None,
         encoding: Optional[SchemaPath] = None,
         **parameters: str,
     ):
         self.style_deserializers_factory = style_deserializers_factory
         self.media_types_deserializer = media_types_deserializer
         self.mimetype = mimetype
         self.schema = schema
+        self.schema_validator = schema_validator
+        self.schema_caster = schema_caster
         self.encoding = encoding
         self.parameters = parameters
 
@@ -86,25 +95,68 @@ def deserialize(self, value: bytes) -> Any:
         ):
             return deserialized
 
-        # decode multipart request bodies
-        return self.decode(deserialized)
+        # decode multipart request bodies if schema provided
+        if self.schema is not None:
+            return self.decode(deserialized)
+        
+        return deserialized
 
     def evolve(
-        self, mimetype: str, schema: Optional[SchemaPath]
+        self,
+        schema: SchemaPath,
+        mimetype: Optional[str] = None,
     ) -> "MediaTypeDeserializer":
         cls = self.__class__
+        
+        schema_validator = None
+        if self.schema_validator is not None:
+            schema_validator = self.schema_validator.evolve(schema)
+        
+        schema_caster = None
+        if self.schema_caster is not None:
+            schema_caster = self.schema_caster.evolve(schema)
 
         return cls(
             self.style_deserializers_factory,
             self.media_types_deserializer,
-            mimetype,
+            mimetype=mimetype or self.mimetype,
             schema=schema,
+            schema_validator=schema_validator,
+            schema_caster=schema_caster,
         )
 
-    def decode(self, location: Mapping[str, Any]) -> Mapping[str, Any]:
+    def decode(self, location: Mapping[str, Any], schema_only: bool = False) -> Mapping[str, Any]:
         # schema is required for multipart
         assert self.schema is not None
         properties = {}
+
+        # For urlencoded/multipart, use caster for oneOf/anyOf detection if validator available
+        if self.schema_validator is not None:
+            one_of_schema = self.schema_validator.get_one_of_schema(
+                location, caster=self.schema_caster
+            )
+            if one_of_schema is not None:
+                one_of_properties = self.evolve(
+                    one_of_schema
+                ).decode(location, schema_only=True)
+                properties.update(one_of_properties)
+
+            any_of_schemas = self.schema_validator.iter_any_of_schemas(
+                location, caster=self.schema_caster
+            )
+            for any_of_schema in any_of_schemas:
+                any_of_properties = self.evolve(
+                    any_of_schema
+                ).decode(location, schema_only=True)
+                properties.update(any_of_properties)
+
+            all_of_schemas = self.schema_validator.iter_all_of_schemas(location)
+            for all_of_schema in all_of_schemas:
+                all_of_properties = self.evolve(
+                    all_of_schema
+                ).decode(location, schema_only=True)
+                properties.update(all_of_properties)
+
         for prop_name, prop_schema in get_properties(self.schema).items():
             try:
                 properties[prop_name] = self.decode_property(
@@ -115,6 +167,9 @@ def decode(self, location: Mapping[str, Any]) -> Mapping[str, Any]:
                     continue
                 properties[prop_name] = prop_schema["default"]
 
+        if schema_only:
+            return properties
+
         return properties
 
     def decode_property(
@@ -175,8 +230,8 @@ def decode_property_content_type(
     ) -> Any:
         prop_content_type = get_content_type(prop_schema, prop_encoding)
         prop_deserializer = self.evolve(
-            prop_content_type,
             prop_schema,
+            mimetype=prop_content_type,
         )
         prop_schema_type = prop_schema.getkey("type", "")
         if (

openapi_core/deserializing/media_types/factories.py

@@ -17,6 +17,7 @@
 from openapi_core.deserializing.styles.factories import (
     StyleDeserializersFactory,
 )
+from openapi_core.validation.schemas.validators import SchemaValidator
 
 
 class MediaTypeDeserializersFactory:
@@ -59,6 +60,7 @@ def create(
         self,
         mimetype: str,
         schema: Optional[SchemaPath] = None,
+        schema_validator: Optional[SchemaValidator] = None,
         parameters: Optional[Mapping[str, str]] = None,
         encoding: Optional[SchemaPath] = None,
         extra_media_type_deserializers: Optional[
@@ -73,12 +75,29 @@ def create(
             self.media_type_deserializers,
             extra_media_type_deserializers,
         )
+        
+        # Create schema caster for urlencoded/multipart content types
+        # Only create if both schema and schema_validator are provided
+        schema_caster = None
+        if (
+            schema is not None
+            and schema_validator is not None
+            and (
+                mimetype == "application/x-www-form-urlencoded"
+                or mimetype.startswith("multipart")
+            )
+        ):
+            schema_caster = self.style_deserializers_factory.schema_casters_factory.create(
+                schema
+            )
 
         return MediaTypeDeserializer(
             self.style_deserializers_factory,
             media_types_deserializer,
             mimetype,
             schema=schema,
+            schema_validator=schema_validator,
+            schema_caster=schema_caster,
             encoding=encoding,
             **parameters,
         )

openapi_core/validation/schemas/validators.py

@@ -1,6 +1,8 @@
 import logging
+from collections.abc import Mapping
 from functools import cached_property
 from functools import partial
+from typing import TYPE_CHECKING
 from typing import Any
 from typing import Iterator
 from typing import Optional
@@ -13,6 +15,9 @@
 from openapi_core.validation.schemas.exceptions import InvalidSchemaValue
 from openapi_core.validation.schemas.exceptions import ValidateError
 
+if TYPE_CHECKING:
+    from openapi_core.casting.schemas.casters import SchemaCaster
+
 log = logging.getLogger(__name__)
 
 
@@ -113,15 +118,39 @@ def iter_valid_schemas(self, value: Any) -> Iterator[SchemaPath]:
     def get_one_of_schema(
         self,
         value: Any,
+        caster: Optional["SchemaCaster"] = None,
     ) -> Optional[SchemaPath]:
+        """Find the matching oneOf schema.
+        
+        Args:
+            value: The value to match against schemas
+            caster: Optional caster for type coercion during matching.
+                    Useful for form-encoded data where types need casting.
+        """
         if "oneOf" not in self.schema:
             return None
 
         one_of_schemas = self.schema / "oneOf"
         for subschema in one_of_schemas:
             validator = self.evolve(subschema)
             try:
-                validator.validate(value)
+                test_value = value
+                # Only cast if caster provided (opt-in behavior)
+                if caster is not None:
+                    try:
+                        # Convert to dict if it's not exactly a plain dict
+                        # (e.g., ImmutableMultiDict from werkzeug)
+                        if type(value) is not dict:
+                            test_value = dict(value)
+                        else:
+                            test_value = value
+                        test_value = caster.evolve(subschema).cast(test_value)
+                    except (ValueError, TypeError, Exception):
+                        # If casting fails, try validation with original value
+                        # We catch generic Exception to handle CastError without circular import
+                        test_value = value
+                
+                validator.validate(test_value)
             except ValidateError:
                 continue
             else:
@@ -133,15 +162,38 @@ def get_one_of_schema(
     def iter_any_of_schemas(
         self,
         value: Any,
+        caster: Optional["SchemaCaster"] = None,
     ) -> Iterator[SchemaPath]:
+        """Iterate matching anyOf schemas.
+        
+        Args:
+            value: The value to match against schemas
+            caster: Optional caster for type coercion during matching.
+                    Useful for form-encoded data where types need casting.
+        """
         if "anyOf" not in self.schema:
             return
 
         any_of_schemas = self.schema / "anyOf"
         for subschema in any_of_schemas:
             validator = self.evolve(subschema)
             try:
-                validator.validate(value)
+                test_value = value
+                # Only cast if caster provided (opt-in behavior)
+                if caster is not None:
+                    try:
+                        # Convert to dict if it's not exactly a plain dict
+                        if type(value) is not dict:
+                            test_value = dict(value)
+                        else:
+                            test_value = value
+                        test_value = caster.evolve(subschema).cast(test_value)
+                    except (ValueError, TypeError, Exception):
+                        # If casting fails, try validation with original value
+                        # We catch generic Exception to handle CastError without circular import
+                        test_value = value
+                
+                validator.validate(test_value)
             except ValidateError:
                 continue
             else:

openapi_core/validation/validators.py

@@ -134,9 +134,17 @@ def _deserialise_media_type(
         encoding = None
         if "encoding" in media_type:
             encoding = media_type.get("encoding")
+        schema_validator = None
+        if schema is not None:
+            schema_validator = self.schema_validators_factory.create(
+                schema,
+                format_validators=self.format_validators,
+                extra_format_validators=self.extra_format_validators,
+            )
         deserializer = self.media_type_deserializers_factory.create(
             mimetype,
             schema=schema,
+            schema_validator=schema_validator,
             parameters=parameters,
             encoding=encoding,
             extra_media_type_deserializers=self.extra_media_type_deserializers,