Validation context

Author: p1c2u

openapi_core/deserializing/media_types/deserializers.py

@@ -24,6 +24,7 @@
 from openapi_core.schema.protocols import SuportsGetList
 from openapi_core.schema.schemas import get_properties
 from openapi_core.validation.schemas.validators import SchemaValidator
+from openapi_core.validation.schemas.exceptions import ValidateError
 
 if TYPE_CHECKING:
     from openapi_core.casting.schemas.casters import SchemaCaster
@@ -126,6 +127,8 @@ def evolve(
             schema=schema,
             schema_validator=schema_validator,
             schema_caster=schema_caster,
+            encoding=self.encoding,
+            **self.parameters,
         )
 
     def decode(
@@ -137,27 +140,21 @@ def decode(
 
         # 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
-            )
+            one_of_schema = self.get_composed_one_of_schema(location)
             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
-            )
+            any_of_schemas = self.iter_composed_any_of_schemas(location)
             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
-            )
+            all_of_schemas = self.iter_composed_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
@@ -253,3 +250,76 @@ def decode_property_content_type(
                 return list(map(prop_deserializer.deserialize, value))
 
         return prop_deserializer.deserialize(location[prop_name])
+
+    def get_composed_one_of_schema(
+        self, location: Mapping[str, Any]
+    ) -> Optional[SchemaPath]:
+        assert self.schema_validator is not None
+
+        if not self.mimetype.startswith("multipart"):
+            return self.schema_validator.get_one_of_schema(
+                location, caster=self.schema_caster
+            )
+
+        if self.schema is None or "oneOf" not in self.schema:
+            return None
+
+        for subschema in self.schema / "oneOf":
+            if self.is_decoded_subschema_valid(subschema, location):
+                return subschema
+
+        return None
+
+    def iter_composed_any_of_schemas(
+        self, location: Mapping[str, Any]
+    ) -> list[SchemaPath]:
+        assert self.schema_validator is not None
+
+        if not self.mimetype.startswith("multipart"):
+            return list(
+                self.schema_validator.iter_any_of_schemas(
+                    location, caster=self.schema_caster
+                )
+            )
+
+        if self.schema is None or "anyOf" not in self.schema:
+            return []
+
+        return [
+            subschema
+            for subschema in self.schema / "anyOf"
+            if self.is_decoded_subschema_valid(subschema, location)
+        ]
+
+    def iter_composed_all_of_schemas(
+        self, location: Mapping[str, Any]
+    ) -> list[SchemaPath]:
+        assert self.schema_validator is not None
+
+        if not self.mimetype.startswith("multipart"):
+            return list(self.schema_validator.iter_all_of_schemas(location))
+
+        if self.schema is None or "allOf" not in self.schema:
+            return []
+
+        return [
+            subschema
+            for subschema in self.schema / "allOf"
+            if self.is_decoded_subschema_valid(subschema, location)
+        ]
+
+    def is_decoded_subschema_valid(
+        self,
+        subschema: SchemaPath,
+        location: Mapping[str, Any],
+    ) -> bool:
+        assert self.schema_validator is not None
+
+        deserializer = self.evolve(subschema)
+        candidate = deserializer.decode(location)
+        validator = self.schema_validator.evolve(subschema)
+        try:
+            validator.validate(candidate)
+        except ValidateError:
+            return False
+        return True

openapi_core/deserializing/styles/util.py

@@ -22,6 +22,19 @@ def split(value: str, separator: str = ",", step: int = 1) -> List[str]:
     return result
 
 
+def decode_form_value(value: Any) -> Any:
+    if isinstance(value, bytes):
+        try:
+            return value.decode("utf-8")
+        except UnicodeDecodeError:
+            return value.decode("ASCII", errors="surrogateescape")
+
+    if isinstance(value, list):
+        return [decode_form_value(item) for item in value]
+
+    return value
+
+
 def delimited_loads(
     explode: bool,
     name: str,
@@ -118,18 +131,19 @@ def form_loads(
     explode_type = (explode, schema_type)
     # color=blue,black,brown
     if explode_type == (False, "array"):
-        return split(location[name], separator=",")
+        value = decode_form_value(location[name])
+        return split(value, separator=",")
     # color=blue&color=black&color=brown
     elif explode_type == (True, "array"):
         if name not in location:
             raise KeyError(name)
         if isinstance(location, SuportsGetAll):
-            return location.getall(name)
+            return decode_form_value(location.getall(name))
         if isinstance(location, SuportsGetList):
-            return location.getlist(name)
-        return location[name]
+            return decode_form_value(location.getlist(name))
+        return decode_form_value(location[name])
 
-    value = location[name]
+    value = decode_form_value(location[name])
     # color=R,100,G,200,B,150
     if explode_type == (False, "object"):
         return dict(map(split, split(value, separator=",", step=2)))

openapi_core/unmarshalling/schemas/unmarshallers.py

@@ -15,6 +15,7 @@
 from openapi_core.unmarshalling.schemas.datatypes import (
     FormatUnmarshallersDict,
 )
+from openapi_core.validation.schemas.datatypes import ValidationState
 from openapi_core.validation.schemas.validators import SchemaValidator
 
 log = logging.getLogger(__name__)
@@ -39,6 +40,14 @@ class ArrayUnmarshaller(PrimitiveUnmarshaller):
     def __call__(self, value: Any) -> Optional[List[Any]]:
         return list(map(self.items_unmarshaller.unmarshal, value))
 
+    def unmarshal_state(self, state: ValidationState) -> Optional[List[Any]]:
+        if state.item_states:
+            return [
+                self.items_unmarshaller.unmarshal_state(item_state)
+                for item_state in state.item_states
+            ]
+        return self(value=state.value)
+
     @property
     def items_unmarshaller(self) -> "SchemaUnmarshaller":
         # sometimes we don't have any schema i.e. free-form objects
@@ -55,6 +64,14 @@ def __call__(self, value: Any) -> Any:
 
         return object_class(**properties)
 
+    def unmarshal_state(self, state: ValidationState) -> Any:
+        properties = self._unmarshal_properties_from_state(state)
+
+        fields: Iterable[str] = properties and properties.keys() or []
+        object_class = self.object_class_factory.create(self.schema, fields)
+
+        return object_class(**properties)
+
     @property
     def object_class_factory(self) -> ModelPathFactory:
         return ModelPathFactory()
@@ -125,6 +142,94 @@ def _unmarshal_properties(
             for prop_name, prop_value in value.items():
                 if prop_name in properties:
                     continue
+                child_state = state.additional_property_states.get(prop_name)
+                if child_state is not None:
+                    properties[prop_name] = (
+                        additional_prop_unmarshaler.unmarshal_state(child_state)
+                    )
+                    continue
+                properties[prop_name] = additional_prop_unmarshaler.unmarshal(
+                    prop_value
+                )
+
+        return properties
+
+    def _unmarshal_properties_from_state(
+        self,
+        state: ValidationState,
+        schema_only: bool = False,
+    ) -> Any:
+        value = state.value
+        properties = {}
+
+        if state.one_of_state is not None:
+            one_of_properties = self.evolve(
+                state.one_of_state.schema
+            )._unmarshal_properties_from_state(
+                state.one_of_state,
+                schema_only=True,
+            )
+            properties.update(one_of_properties)
+
+        for any_of_state in state.any_of_states:
+            any_of_properties = self.evolve(
+                any_of_state.schema
+            )._unmarshal_properties_from_state(
+                any_of_state,
+                schema_only=True,
+            )
+            properties.update(any_of_properties)
+
+        for all_of_state in state.all_of_states:
+            all_of_properties = self.evolve(
+                all_of_state.schema
+            )._unmarshal_properties_from_state(
+                all_of_state,
+                schema_only=True,
+            )
+            properties.update(all_of_properties)
+
+        for prop_name, prop_schema in get_properties(self.schema).items():
+            child_state = state.property_states.get(prop_name)
+            if child_state is not None:
+                properties[prop_name] = self.schema_unmarshaller.evolve(
+                    prop_schema
+                ).unmarshal_state(child_state)
+                continue
+
+            try:
+                prop_value = value[prop_name]
+            except KeyError:
+                if "default" not in prop_schema:
+                    continue
+                prop_value = (prop_schema / "default").read_value()
+            properties[prop_name] = self.schema_unmarshaller.evolve(
+                prop_schema
+            ).unmarshal(prop_value)
+
+        if schema_only:
+            return properties
+
+        additional_properties = self.schema.get("additionalProperties", True)
+        if additional_properties is not False:
+            if additional_properties is True:
+                additional_prop_schema = SchemaPath.from_dict(
+                    {"nullable": True}
+                )
+            else:
+                additional_prop_schema = self.schema / "additionalProperties"
+            additional_prop_unmarshaler = self.schema_unmarshaller.evolve(
+                additional_prop_schema
+            )
+            for prop_name, prop_value in value.items():
+                if prop_name in properties:
+                    continue
+                child_state = state.additional_property_states.get(prop_name)
+                if child_state is not None:
+                    properties[prop_name] = (
+                        additional_prop_unmarshaler.unmarshal_state(child_state)
+                    )
+                    continue
                 properties[prop_name] = additional_prop_unmarshaler.unmarshal(
                     prop_value
                 )
@@ -143,6 +248,17 @@ def __call__(self, value: Any) -> Any:
         )
         return unmarshaller(value)
 
+    def unmarshal_state(self, state: ValidationState) -> Any:
+        primitive_type = state.primitive_type
+        if primitive_type is None:
+            return None
+        unmarshaller = self.schema_unmarshaller.get_type_unmarshaller(
+            primitive_type
+        )
+        if hasattr(unmarshaller, "unmarshal_state"):
+            return unmarshaller.unmarshal_state(state)
+        return unmarshaller(state.value)
+
 
 class AnyUnmarshaller(MultiTypeUnmarshaller):
     pass
@@ -239,7 +355,11 @@ def __init__(
         self.formats_unmarshaller = formats_unmarshaller
 
     def unmarshal(self, value: Any) -> Any:
-        self.schema_validator.validate(value)
+        state = self.schema_validator.validate_state(value)
+        return self.unmarshal_state(state)
+
+    def unmarshal_state(self, state: ValidationState) -> Any:
+        value = state.value
 
         # skip unmarshalling for nullable in OpenAPI 3.0
         if value is None and (self.schema / "nullable").read_bool(
@@ -249,11 +369,14 @@ def unmarshal(self, value: Any) -> Any:
 
         schema_type = (self.schema / "type").read_str_or_list(None)
         type_unmarshaller = self.get_type_unmarshaller(schema_type)
-        typed = type_unmarshaller(value)
+        if hasattr(type_unmarshaller, "unmarshal_state"):
+            typed = type_unmarshaller.unmarshal_state(state)
+        else:
+            typed = type_unmarshaller(value)
         # skip finding format for None
         if typed is None:
             return None
-        schema_format = self.find_format(value)
+        schema_format = self.find_format(value, state=state)
         if schema_format is None:
             return typed
         # ignore incompatible formats
@@ -300,7 +423,21 @@ def evolve(self, schema: SchemaPath) -> "SchemaUnmarshaller":
             self.formats_unmarshaller,
         )
 
-    def find_format(self, value: Any) -> Optional[str]:
+    def find_format(
+        self,
+        value: Any,
+        state: Optional[ValidationState] = None,
+    ) -> Optional[str]:
+        if state is not None:
+            for schema in self.iter_valid_schemas_from_state(state):
+                schema_validator = self.schema_validator.evolve(schema)
+                primitive_type = schema_validator.get_primitive_type(value)
+                if primitive_type != "string":
+                    continue
+                if "format" in schema:
+                    return (schema / "format").read_str()
+            return None
+
         for schema in self.schema_validator.iter_valid_schemas(value):
             schema_validator = self.schema_validator.evolve(schema)
             primitive_type = schema_validator.get_primitive_type(value)
@@ -309,3 +446,18 @@ def find_format(self, value: Any) -> Optional[str]:
             if "format" in schema:
                 return (schema / "format").read_str()
         return None
+
+    def iter_valid_schemas_from_state(
+        self,
+        state: ValidationState,
+    ) -> Iterable[SchemaPath]:
+        yield state.schema
+
+        if state.one_of_state is not None:
+            yield state.one_of_state.schema
+
+        for any_of_state in state.any_of_states:
+            yield any_of_state.schema
+
+        for all_of_state in state.all_of_states:
+            yield all_of_state.schema