Media type encoding support

Author: p1c2u

Diff for: openapi_core/deserializing/media_types/deserializers.py

@@ -1,31 +1,113 @@
 import warnings
 from typing import Any
+from typing import Dict
 from typing import Optional
 from xml.etree.ElementTree import ParseError
 
 from openapi_core.deserializing.media_types.datatypes import (
     DeserializerCallable,
 )
+from openapi_core.deserializing.media_types.datatypes import (
+    MediaTypeDeserializersDict,
+)
 from openapi_core.deserializing.media_types.exceptions import (
     MediaTypeDeserializeError,
 )
+from openapi_core.schema.encodings import get_encoding_default_content_type
+from openapi_core.spec import Spec
 
 
-class CallableMediaTypeDeserializer:
+class ContentTypesDeserializer:
     def __init__(
         self,
-        mimetype: str,
-        deserializer_callable: Optional[DeserializerCallable] = None,
+        media_type_deserializers: Optional[MediaTypeDeserializersDict] = None,
+        extra_media_type_deserializers: Optional[MediaTypeDeserializersDict] = None,
     ):
-        self.mimetype = mimetype
-        self.deserializer_callable = deserializer_callable
+        if media_type_deserializers is None:
+            media_type_deserializers = {}
+        self.media_type_deserializers = media_type_deserializers
+        if extra_media_type_deserializers is None:
+            extra_media_type_deserializers = {}
+        self.extra_media_type_deserializers = extra_media_type_deserializers
 
-    def deserialize(self, value: Any) -> Any:
-        if self.deserializer_callable is None:
-            warnings.warn(f"Unsupported {self.mimetype} mimetype")
+    def deserialize(self, mimetype: str, value: Any) -> Any:
+        deserializer_callable = self.get_deserializer_callable(mimetype)
+        if deserializer_callable is None:
+            warnings.warn(f"Unsupported {mimetype} mimetype")
             return value
 
         try:
-            return self.deserializer_callable(value)
+            return deserializer_callable(value)
         except (ParseError, ValueError, TypeError, AttributeError):
-            raise MediaTypeDeserializeError(self.mimetype, value)
+            raise MediaTypeDeserializeError(mimetype, value)
+
+    def get_deserializer_callable(
+        self,
+        mimetype: str,
+    ) -> Optional[DeserializerCallable]:
+        if mimetype in self.extra_media_type_deserializers:
+            return self.extra_media_type_deserializers[mimetype]
+        return self.media_type_deserializers.get(mimetype)
+
+
+class MediaTypeDeserializer(ContentTypesDeserializer):
+    def __init__(
+        self,
+        schema: Spec,
+        mimetype: str,
+        content_types_deserializers: ContentTypesDeserializer,
+        encoding: Optional[Spec] = None,
+    ):
+        self.schema = schema
+        self.mimetype = mimetype
+        self.content_types_deserializers = content_types_deserializers
+        self.encoding = encoding
+
+    def deserialize(self, value: Any) -> Any:
+        deserialized = self.content_types_deserializers.deserialize(self.mimetype, value)
+
+        if self.mimetype != "application/x-www-form-urlencoded" and not self.mimetype.startswith("multipart"):
+            return deserialized
+
+        return self.decode(deserialized)
+
+    def evolve(self, schema: Spec, mimetype: str) -> "MediaTypeDeserializer":
+        cls = self.__class__
+
+        return cls(
+            schema,
+            mimetype,
+            self.content_types_deserializers,
+        )
+
+    def decode(self, value: Dict[str, Any]) -> Dict[str, Any]:
+        return {
+            prop_name: self.decode_property(prop_name, prop_value)
+            for prop_name, prop_value in value.items()
+        }
+
+    def decode_property(self, prop_name: str, value: Any) -> Any:
+        schema_props = self.schema.get("properties")
+        prop_schema = None
+        if schema_props is not None and prop_name in schema_props:
+            prop_schema = self.schema.get(prop_name)
+        prop_content_type = self.get_property_content_type(prop_name, prop_schema)
+        prop_deserializer = self.evolve(
+            schema=prop_schema,
+            mimetype=prop_content_type,
+        )
+        return prop_deserializer.deserialize(value)
+
+    def get_property_content_type(self, prop_name: str, prop_schema: Optional[Spec] = None) -> str:
+        if self.encoding is None:
+            return get_encoding_default_content_type(prop_schema)
+
+        if prop_name not in self.encoding:
+            return get_encoding_default_content_type(prop_schema)
+
+        prep_encoding = self.encoding.get(prop_name)
+        prop_content_type = prep_encoding.getkey("contentType")
+        if prop_content_type is None:
+            return get_encoding_default_content_type(prop_schema)
+
+        return prop_content_type

Diff for: openapi_core/deserializing/media_types/factories.py

@@ -7,8 +7,12 @@
     MediaTypeDeserializersDict,
 )
 from openapi_core.deserializing.media_types.deserializers import (
-    CallableMediaTypeDeserializer,
+    ContentTypesDeserializer,
 )
+from openapi_core.deserializing.media_types.deserializers import (
+    MediaTypeDeserializer,
+)
+from openapi_core.spec import Spec
 
 
 class MediaTypeDeserializersFactory:
@@ -22,25 +26,23 @@ def __init__(
 
     def create(
         self,
+        schema: Spec,
         mimetype: str,
+        encoding: Optional[Spec] = None,
         extra_media_type_deserializers: Optional[
             MediaTypeDeserializersDict
         ] = None,
-    ) -> CallableMediaTypeDeserializer:
+    ) -> MediaTypeDeserializer:
         if extra_media_type_deserializers is None:
             extra_media_type_deserializers = {}
-        deserialize_callable = self.get_deserializer_callable(
-            mimetype,
-            extra_media_type_deserializers=extra_media_type_deserializers,
-        )
 
-        return CallableMediaTypeDeserializer(mimetype, deserialize_callable)
+        content_types_deserializer = ContentTypesDeserializer(
+            self.media_type_deserializers,
+            extra_media_type_deserializers,
+        )
 
-    def get_deserializer_callable(
-        self,
-        mimetype: str,
-        extra_media_type_deserializers: MediaTypeDeserializersDict,
-    ) -> Optional[DeserializerCallable]:
-        if mimetype in extra_media_type_deserializers:
-            return extra_media_type_deserializers[mimetype]
-        return self.media_type_deserializers.get(mimetype)
+        return MediaTypeDeserializer(
+            schema, mimetype,
+            content_types_deserializer,
+            encoding=encoding,
+        )

Diff for: openapi_core/schema/encodings.py

@@ -0,0 +1,27 @@
+from typing import Optional
+
+from openapi_core.spec import Spec
+
+
+def get_encoding_default_content_type(prop_schema: Optional[Spec]) -> str:
+    if prop_schema is None:
+        return "text/plain"
+
+    prop_type = prop_schema.getkey("type")
+    if prop_type is None:
+        return "text/plain"
+
+    prop_format = prop_schema.getkey("format")
+    if prop_type == "string" and prop_format in ["binary", "base64"]:
+        return "application/octet-stream"
+
+    if prop_type == "object":
+        return "application/json"
+
+    if prop_type == "array":
+        prop_items = prop_schema.get("items")
+        prop_items_type = prop_items.getkey("type")
+        if prop_items_type == "object":
+            return "application/json"
+
+    return "text/plain"

Diff for: openapi_core/validation/validators.py

@@ -78,9 +78,16 @@ def _get_media_type(self, content: Spec, mimetype: str) -> MediaType:
         finder = MediaTypeFinder(content)
         return finder.find(mimetype)
 
-    def _deserialise_data(self, mimetype: str, value: Any) -> Any:
+    def _deserialise_data(self, media_type: Spec, mimetype: str, value: Any) -> Any:
+        schema = media_type.get("schema")
+        encoding = None
+        if "encoding" in media_type:
+            encoding = media_type.get("encoding")
+
         deserializer = self.media_type_deserializers_factory.create(
+            schema,
             mimetype,
+            encoding=encoding,
             extra_media_type_deserializers=self.extra_media_type_deserializers,
         )
         return deserializer.deserialize(value)
@@ -152,7 +159,7 @@ def _get_param_or_header_value_and_schema(
             else:
                 content = param_or_header / "content"
                 mimetype, media_type = next(content.items())
-                deserialised = self._deserialise_data(mimetype, raw_value)
+                deserialised = self._deserialise_data(media_type, mimetype, raw_value)
                 schema = media_type / "schema"
             casted = self._cast(schema, deserialised)
         return casted, schema
@@ -161,7 +168,7 @@ def _get_content_value_and_schema(
         self, raw: Any, mimetype: str, content: Spec
     ) -> Tuple[Any, Optional[Spec]]:
         media_type, mimetype = self._get_media_type(content, mimetype)
-        deserialised = self._deserialise_data(mimetype, raw)
+        deserialised = self._deserialise_data(media_type, mimetype, raw)
         casted = self._cast(media_type, deserialised)
 
         if "schema" not in media_type:

Diff for: tests/unit/deserializing/test_media_types_deserializers.py

@@ -14,13 +14,19 @@ class TestMediaTypeDeserializer:
     def deserializer_factory(self):
         def create_deserializer(
             media_type,
+            schema=None,
+            encoding=None,
             media_type_deserializers=media_type_deserializers,
             extra_media_type_deserializers=None,
         ):
+            if schema is None:
+                schema or {}
             return MediaTypeDeserializersFactory(
                 media_type_deserializers,
             ).create(
+                schema,
                 media_type,
+                encoding=encoding,
                 extra_media_type_deserializers=extra_media_type_deserializers,
             )
 
@@ -132,7 +138,15 @@ def test_urlencoded_form_empty(self, deserializer_factory):
 
     def test_urlencoded_form_simple(self, deserializer_factory):
         mimetype = "application/x-www-form-urlencoded"
-        deserializer = deserializer_factory(mimetype)
+        schema = {
+            "type": "object",
+            "properties": {
+                "param1": {
+                    "type": "string",
+                },
+            },
+        }
+        deserializer = deserializer_factory(mimetype, schema=schema)
         value = "param1=test"
 
         result = deserializer.deserialize(value)
@@ -150,7 +164,16 @@ def test_data_form_empty(self, deserializer_factory, value):
 
     def test_data_form_simple(self, deserializer_factory):
         mimetype = "multipart/form-data"
-        deserializer = deserializer_factory(mimetype)
+        schema = {
+            "type": "object",
+            "properties": {
+                "param1": {
+                    "type": "string",
+                    "format": "binary",
+                },
+            },
+        }
+        deserializer = deserializer_factory(mimetype, schema=schema)
         value = (
             b'Content-Type: multipart/form-data; boundary="'
             b'===============2872712225071193122=="\n'