Merge pull request #1181 from litteratum/fix_oas31_multitype

fix: OAS 3.1 multi type (cast/deserialize)

Author: p1c2u

openapi_core/casting/schemas/casters.py

@@ -235,7 +235,7 @@ def cast(self, value: Any) -> Any:
         ):
             return value
 
-        schema_type = (self.schema / "type").read_str(None)
+        schema_type = (self.schema / "type").read_str_or_list(None)
         type_caster = self.get_type_caster(schema_type)
 
         if value is None:

openapi_core/casting/schemas/exceptions.py

@@ -9,7 +9,7 @@ class CastError(DeserializeError):
     """Schema cast operation error"""
 
     value: Any
-    type: str | None
+    type: str | list[str] | None
 
     def __str__(self) -> str:
         return f"Failed to cast value to {self.type} type: {self.value}"

openapi_core/deserializing/styles/datatypes.py

@@ -3,5 +3,7 @@
 from typing import Dict
 from typing import Mapping
 
-DeserializerCallable = Callable[[bool, str, str, Mapping[str, Any]], Any]
+DeserializerCallable = Callable[
+    [bool, str, str | list[str], Mapping[str, Any]], Any
+]
 StyleDeserializersDict = Dict[str, DeserializerCallable]

openapi_core/deserializing/styles/deserializers.py

@@ -15,7 +15,7 @@ def __init__(
         style: str,
         explode: bool,
         name: str,
-        schema_type: str,
+        schema_type: str | list[str],
         caster: SchemaCaster,
         deserializer_callable: Optional[DeserializerCallable] = None,
     ):

openapi_core/deserializing/styles/factories.py

@@ -28,7 +28,7 @@ def create(
     ) -> StyleDeserializer:
         deserialize_callable = self.style_deserializers.get(style)
         caster = self.schema_casters_factory.create(spec, schema)
-        schema_type = (schema / "type").read_str("")
+        schema_type = (schema / "type").read_str_or_list("")
         return StyleDeserializer(
             style, explode, name, schema_type, caster, deserialize_callable
         )

openapi_core/deserializing/styles/util.py

@@ -25,7 +25,7 @@ def split(value: str, separator: str = ",", step: int = 1) -> List[str]:
 def delimited_loads(
     explode: bool,
     name: str,
-    schema_type: str,
+    schema_type: str | list[str],
     location: Mapping[str, Any],
     delimiter: str,
 ) -> Any:
@@ -46,7 +46,10 @@ def delimited_loads(
 
 
 def matrix_loads(
-    explode: bool, name: str, schema_type: str, location: Mapping[str, Any]
+    explode: bool,
+    name: str,
+    schema_type: str | list[str],
+    location: Mapping[str, Any],
 ) -> Any:
     if explode == False:
         m = re.match(rf"^;{name}=(.*)$", location[f";{name}"])
@@ -83,7 +86,10 @@ def matrix_loads(
 
 
 def label_loads(
-    explode: bool, name: str, schema_type: str, location: Mapping[str, Any]
+    explode: bool,
+    name: str,
+    schema_type: str | list[str],
+    location: Mapping[str, Any],
 ) -> Any:
     if explode == False:
         value = location[f".{name}"]
@@ -113,7 +119,10 @@ def label_loads(
 
 
 def form_loads(
-    explode: bool, name: str, schema_type: str, location: Mapping[str, Any]
+    explode: bool,
+    name: str,
+    schema_type: str | list[str],
+    location: Mapping[str, Any],
 ) -> Any:
     explode_type = (explode, schema_type)
     # color=blue,black,brown
@@ -144,7 +153,10 @@ def form_loads(
 
 
 def simple_loads(
-    explode: bool, name: str, schema_type: str, location: Mapping[str, Any]
+    explode: bool,
+    name: str,
+    schema_type: str | list[str],
+    location: Mapping[str, Any],
 ) -> Any:
     value = location[name]
 
@@ -167,21 +179,30 @@ def simple_loads(
 
 
 def space_delimited_loads(
-    explode: bool, name: str, schema_type: str, location: Mapping[str, Any]
+    explode: bool,
+    name: str,
+    schema_type: str | list[str],
+    location: Mapping[str, Any],
 ) -> Any:
     return delimited_loads(
         explode, name, schema_type, location, delimiter="%20"
     )
 
 
 def pipe_delimited_loads(
-    explode: bool, name: str, schema_type: str, location: Mapping[str, Any]
+    explode: bool,
+    name: str,
+    schema_type: str | list[str],
+    location: Mapping[str, Any],
 ) -> Any:
     return delimited_loads(explode, name, schema_type, location, delimiter="|")
 
 
 def deep_object_loads(
-    explode: bool, name: str, schema_type: str, location: Mapping[str, Any]
+    explode: bool,
+    name: str,
+    schema_type: str | list[str],
+    location: Mapping[str, Any],
 ) -> Any:
     explode_type = (explode, schema_type)
 

tests/unit/casting/test_schema_casters.py

@@ -67,6 +67,26 @@ def test_array_invalid_value(self, value, caster_factory):
         ):
             caster_factory(schema).cast(value)
 
+    @pytest.mark.parametrize(
+        "schema_types,value",
+        [
+            (["string", "number", "boolean"], "12567"),
+            (["integer", "string"], "42"),
+            (["number", "string"], "3.14"),
+            (["boolean", "string"], "true"),
+        ],
+    )
+    def test_oas31_multi_type(self, caster_factory, schema_types, value):
+        """Test OAS 3.1 list-style `type`."""
+        spec = {
+            "type": schema_types,
+        }
+        schema = SchemaPath.from_dict(spec)
+
+        result = caster_factory(schema).cast(value)
+
+        assert result == value
+
     @pytest.mark.parametrize(
         "composite_type,schema_type,value,expected",
         [

tests/unit/deserializing/test_styles_deserializers.py

@@ -444,6 +444,34 @@ def test_pipe_delimited_valid(
 
         assert result == expected
 
+    @pytest.mark.parametrize(
+        "schema_types,value,expected",
+        [
+            (["string", "number", "boolean"], "12567", "12567"),
+            (["integer", "string"], "42", "42"),
+        ],
+    )
+    def test_oas31_multi_type_form(
+        self, deserializer_factory, schema_types, value, expected
+    ):
+        """Test OAS 3.1 multi-type support for form style parameters."""
+        name = "param"
+        spec = {
+            "name": name,
+            "in": "query",
+            "explode": True,
+            "schema": {
+                "type": schema_types,
+            },
+        }
+        param = SchemaPath.from_dict(spec)
+        deserializer = deserializer_factory(param)
+        location = {name: value}
+
+        result = deserializer.deserialize(location)
+
+        assert result == expected
+
     def test_deep_object_valid(self, deserializer_factory):
         name = "param"
         spec = {