Mimetype parameters handling

Author: p1c2u

openapi_core/deserializing/media_types/deserializers.py

@@ -16,16 +16,18 @@ def __init__(
         self,
         mimetype: str,
         deserializer_callable: Optional[DeserializerCallable] = None,
+        **parameters: str,
     ):
         self.mimetype = mimetype
         self.deserializer_callable = deserializer_callable
+        self.parameters = parameters
 
     def deserialize(self, value: Any) -> Any:
         if self.deserializer_callable is None:
             warnings.warn(f"Unsupported {self.mimetype} mimetype")
             return value
 
         try:
-            return self.deserializer_callable(value)
+            return self.deserializer_callable(value, **self.parameters)
         except (ParseError, ValueError, TypeError, AttributeError):
             raise MediaTypeDeserializeError(self.mimetype, value)

openapi_core/deserializing/media_types/factories.py

@@ -1,3 +1,4 @@
+from typing import Mapping
 from typing import Optional
 
 from openapi_core.deserializing.media_types.datatypes import (
@@ -23,6 +24,7 @@ def __init__(
     def create(
         self,
         mimetype: str,
+        parameters: Mapping[str, str],
         extra_media_type_deserializers: Optional[
             MediaTypeDeserializersDict
         ] = None,
@@ -34,7 +36,9 @@ def create(
             extra_media_type_deserializers=extra_media_type_deserializers,
         )
 
-        return CallableMediaTypeDeserializer(mimetype, deserialize_callable)
+        return CallableMediaTypeDeserializer(
+            mimetype, deserialize_callable, **parameters
+        )
 
     def get_deserializer_callable(
         self,

openapi_core/deserializing/media_types/util.py

@@ -5,17 +5,26 @@
 from urllib.parse import parse_qsl
 
 
-def plain_loads(value: Union[str, bytes]) -> str:
+def plain_loads(value: Union[str, bytes], **parameters: str) -> str:
+    charset = "utf-8"
+    if "charset" in parameters:
+        charset = parameters["charset"]
     if isinstance(value, bytes):
-        value = value.decode("ASCII", errors="surrogateescape")
+        try:
+            return value.decode(charset)
+        # fallback safe decode
+        except UnicodeDecodeError:
+            return value.decode("ASCII", errors="surrogateescape")
     return value
 
 
-def urlencoded_form_loads(value: Any) -> Dict[str, Any]:
+def urlencoded_form_loads(value: Any, **parameters: str) -> Dict[str, Any]:
     return dict(parse_qsl(value))
 
 
-def data_form_loads(value: Union[str, bytes]) -> Dict[str, Any]:
+def data_form_loads(
+    value: Union[str, bytes], **parameters: str
+) -> Dict[str, Any]:
     if isinstance(value, bytes):
         value = value.decode("ASCII", errors="surrogateescape")
     parser = Parser()

openapi_core/templating/media_types/datatypes.py

@@ -1,3 +1,6 @@
 from collections import namedtuple
+from dataclasses import dataclass
+from typing import Mapping
+from typing import Optional
 
-MediaType = namedtuple("MediaType", ["value", "key"])
+MediaType = namedtuple("MediaType", ["mime_type", "parameters", "media_type"])

openapi_core/templating/media_types/finders.py

@@ -1,5 +1,7 @@
 """OpenAPI core templating media types finders module"""
 import fnmatch
+from typing import Mapping
+from typing import Tuple
 
 from openapi_core.spec import Spec
 from openapi_core.templating.media_types.datatypes import MediaType
@@ -12,15 +14,31 @@ def __init__(self, content: Spec):
 
     def get_first(self) -> MediaType:
         mimetype, media_type = next(self.content.items())
-        return MediaType(media_type, mimetype)
+        return MediaType(mimetype, {}, media_type)
 
     def find(self, mimetype: str) -> MediaType:
-        if mimetype in self.content:
-            return MediaType(self.content / mimetype, mimetype)
+        mime_type, parameters = self._parse_mimetype(mimetype)
 
-        if mimetype:
+        # simple mime type
+        for m in [mimetype, mime_type]:
+            if m in self.content:
+                return MediaType(mime_type, parameters, self.content / m)
+
+        # range mime type
+        if mime_type:
             for key, value in self.content.items():
-                if fnmatch.fnmatch(mimetype, key):
-                    return MediaType(value, key)
+                if fnmatch.fnmatch(mime_type, key):
+                    return MediaType(key, parameters, value)
 
         raise MediaTypeNotFound(mimetype, list(self.content.keys()))
+
+    def _parse_mimetype(self, mimetype: str) -> Tuple[str, Mapping[str, str]]:
+        mimetype_parts = mimetype.split("; ")
+        mime_type = mimetype_parts[0]
+        parameters = {}
+        if len(mimetype_parts) > 1:
+            parameters_list = (
+                param_str.split("=") for param_str in mimetype_parts[1:]
+            )
+            parameters = dict(parameters_list)
+        return mime_type, parameters

openapi_core/validation/validators.py

@@ -86,10 +86,13 @@ def _find_media_type(
             return finder.get_first()
         return finder.find(mimetype)
 
-    def _deserialise_media_type(self, mimetype: str, value: Any) -> Any:
+    def _deserialise_media_type(
+        self, mimetype: str, parameters: Mapping[str, str], value: Any
+    ) -> Any:
         deserializer = self.media_type_deserializers_factory.create(
             mimetype,
             extra_media_type_deserializers=self.extra_media_type_deserializers,
+            parameters=parameters,
         )
         return deserializer.deserialize(value)
 
@@ -194,8 +197,10 @@ def _convert_content_schema_value_and_schema(
         content: Spec,
         mimetype: Optional[str] = None,
     ) -> Tuple[Any, Optional[Spec]]:
-        media_type, mime_type = self._find_media_type(content, mimetype)
-        deserialised = self._deserialise_media_type(mime_type, raw)
+        mime_type, parameters, media_type = self._find_media_type(
+            content, mimetype
+        )
+        deserialised = self._deserialise_media_type(mime_type, parameters, raw)
         casted = self._cast(media_type, deserialised)
 
         if "schema" not in media_type:

tests/integration/test_petstore.py

@@ -231,7 +231,9 @@ def test_get_pets_response_no_schema(self, spec):
         assert result.body is None
 
         data = "<html></html>"
-        response = MockResponse(data, status_code=404, mimetype="text/html")
+        response = MockResponse(
+            data, status_code=404, mimetype="text/html; charset=utf-8"
+        )
 
         response_result = unmarshal_response(request, response, spec=spec)