media type deserializers refactor

Author: p1c2u

docs/customizations.rst

@@ -14,32 +14,26 @@ If you know you have a valid specification already, disabling the validator can
 
    spec = Spec.from_dict(spec_dict, validator=None)
 
-Deserializers
--------------
+Media type deserializers
+------------------------
 
-Pass custom defined media type deserializers dictionary with supported mimetypes as a key to `MediaTypeDeserializersFactory` and then pass it to `RequestValidator` or `ResponseValidator` constructor:
+Pass custom defined media type deserializers dictionary with supported mimetypes as a key to `unmarshal_response` function:
 
 .. code-block:: python
 
-   from openapi_core.deserializing.media_types.factories import MediaTypeDeserializersFactory
-
    def protobuf_deserializer(message):
        feature = route_guide_pb2.Feature()
        feature.ParseFromString(message)
        return feature
 
-   custom_media_type_deserializers = {
+   extra_media_type_deserializers = {
        'application/protobuf': protobuf_deserializer,
    }
-   media_type_deserializers_factory = MediaTypeDeserializersFactory(
-       custom_deserializers=custom_media_type_deserializers,
-   )
 
-   result = validate_response(
+   result = unmarshal_response(
        request, response,
        spec=spec,
-       cls=V30ResponseValidator,
-       media_type_deserializers_factory=media_type_deserializers_factory,
+       extra_media_type_deserializers=extra_media_type_deserializers,
    )
 
 Format validators

openapi_core/deserializing/media_types/__init__.py

@@ -1,7 +1,22 @@
+from json import loads
+
+from openapi_core.deserializing.media_types.datatypes import (
+    MediaTypeDeserializersDict,
+)
 from openapi_core.deserializing.media_types.factories import (
     MediaTypeDeserializersFactory,
 )
+from openapi_core.deserializing.media_types.util import data_form_loads
+from openapi_core.deserializing.media_types.util import urlencoded_form_loads
 
 __all__ = ["media_type_deserializers_factory"]
 
-media_type_deserializers_factory = MediaTypeDeserializersFactory()
+media_type_deserializers: MediaTypeDeserializersDict = {
+    "application/json": loads,
+    "application/x-www-form-urlencoded": urlencoded_form_loads,
+    "multipart/form-data": data_form_loads,
+}
+
+media_type_deserializers_factory = MediaTypeDeserializersFactory(
+    media_type_deserializers=media_type_deserializers,
+)

openapi_core/deserializing/media_types/datatypes.py

@@ -1,4 +1,6 @@
 from typing import Any
 from typing import Callable
+from typing import Dict
 
 DeserializerCallable = Callable[[Any], Any]
+MediaTypeDeserializersDict = Dict[str, DeserializerCallable]

openapi_core/deserializing/media_types/deserializers.py

@@ -1,6 +1,6 @@
 import warnings
 from typing import Any
-from typing import Callable
+from typing import Optional
 
 from openapi_core.deserializing.media_types.datatypes import (
     DeserializerCallable,
@@ -10,28 +10,20 @@
 )
 
 
-class BaseMediaTypeDeserializer:
-    def __init__(self, mimetype: str):
-        self.mimetype = mimetype
-
-    def __call__(self, value: Any) -> Any:
-        raise NotImplementedError
-
-
-class UnsupportedMimetypeDeserializer(BaseMediaTypeDeserializer):
-    def __call__(self, value: Any) -> Any:
-        warnings.warn(f"Unsupported {self.mimetype} mimetype")
-        return value
-
-
-class CallableMediaTypeDeserializer(BaseMediaTypeDeserializer):
+class CallableMediaTypeDeserializer:
     def __init__(
-        self, mimetype: str, deserializer_callable: DeserializerCallable
+        self,
+        mimetype: str,
+        deserializer_callable: Optional[DeserializerCallable] = None,
     ):
         self.mimetype = mimetype
         self.deserializer_callable = deserializer_callable
 
-    def __call__(self, value: Any) -> Any:
+    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)
         except (ValueError, TypeError, AttributeError):

openapi_core/deserializing/media_types/factories.py

@@ -1,51 +1,60 @@
-from json import loads
-from typing import Any
-from typing import Callable
+import warnings
 from typing import Dict
 from typing import Optional
 
 from openapi_core.deserializing.media_types.datatypes import (
     DeserializerCallable,
 )
-from openapi_core.deserializing.media_types.deserializers import (
-    BaseMediaTypeDeserializer,
+from openapi_core.deserializing.media_types.datatypes import (
+    MediaTypeDeserializersDict,
 )
 from openapi_core.deserializing.media_types.deserializers import (
     CallableMediaTypeDeserializer,
 )
-from openapi_core.deserializing.media_types.deserializers import (
-    UnsupportedMimetypeDeserializer,
-)
-from openapi_core.deserializing.media_types.util import data_form_loads
-from openapi_core.deserializing.media_types.util import urlencoded_form_loads
 
 
 class MediaTypeDeserializersFactory:
-    MEDIA_TYPE_DESERIALIZERS: Dict[str, DeserializerCallable] = {
-        "application/json": loads,
-        "application/x-www-form-urlencoded": urlencoded_form_loads,
-        "multipart/form-data": data_form_loads,
-    }
-
     def __init__(
         self,
-        custom_deserializers: Optional[Dict[str, DeserializerCallable]] = None,
+        media_type_deserializers: Optional[MediaTypeDeserializersDict] = None,
+        custom_deserializers: Optional[MediaTypeDeserializersDict] = None,
     ):
+        if media_type_deserializers is None:
+            media_type_deserializers = {}
+        self.media_type_deserializers = media_type_deserializers
         if custom_deserializers is None:
             custom_deserializers = {}
+        else:
+            warnings.warn(
+                "custom_deserializers is deprecated. "
+                "Use extra_media_type_deserializers.",
+                DeprecationWarning,
+            )
         self.custom_deserializers = custom_deserializers
 
-    def create(self, mimetype: str) -> BaseMediaTypeDeserializer:
-        deserialize_callable = self.get_deserializer_callable(mimetype)
-
-        if deserialize_callable is None:
-            return UnsupportedMimetypeDeserializer(mimetype)
+    def create(
+        self,
+        mimetype: str,
+        extra_media_type_deserializers: Optional[
+            MediaTypeDeserializersDict
+        ] = None,
+    ) -> CallableMediaTypeDeserializer:
+        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)
 
     def get_deserializer_callable(
-        self, mimetype: str
+        self,
+        mimetype: str,
+        extra_media_type_deserializers: MediaTypeDeserializersDict,
     ) -> Optional[DeserializerCallable]:
         if mimetype in self.custom_deserializers:
             return self.custom_deserializers[mimetype]
-        return self.MEDIA_TYPE_DESERIALIZERS.get(mimetype)
+        if mimetype in extra_media_type_deserializers:
+            return extra_media_type_deserializers[mimetype]
+        return self.media_type_deserializers.get(mimetype)

openapi_core/deserializing/parameters/deserializers.py

@@ -2,6 +2,7 @@
 from typing import Any
 from typing import Callable
 from typing import List
+from typing import Optional
 
 from openapi_core.deserializing.exceptions import DeserializeError
 from openapi_core.deserializing.parameters.datatypes import (
@@ -15,35 +16,25 @@
 from openapi_core.spec import Spec
 
 
-class BaseParameterDeserializer:
-    def __init__(self, param_or_header: Spec, style: str):
-        self.param_or_header = param_or_header
-        self.style = style
-
-    def __call__(self, value: Any) -> Any:
-        raise NotImplementedError
-
-
-class UnsupportedStyleDeserializer(BaseParameterDeserializer):
-    def __call__(self, value: Any) -> Any:
-        warnings.warn(f"Unsupported {self.style} style")
-        return value
-
-
-class CallableParameterDeserializer(BaseParameterDeserializer):
+class CallableParameterDeserializer:
     def __init__(
         self,
         param_or_header: Spec,
         style: str,
-        deserializer_callable: DeserializerCallable,
+        deserializer_callable: Optional[DeserializerCallable] = None,
     ):
-        super().__init__(param_or_header, style)
+        self.param_or_header = param_or_header
+        self.style = style
         self.deserializer_callable = deserializer_callable
 
         self.aslist = get_aslist(self.param_or_header)
         self.explode = get_explode(self.param_or_header)
 
-    def __call__(self, value: Any) -> Any:
+    def deserialize(self, value: Any) -> Any:
+        if self.deserializer_callable is None:
+            warnings.warn(f"Unsupported {self.style} style")
+            return value
+
         # if "in" not defined then it's a Header
         if "allowEmptyValue" in self.param_or_header:
             warnings.warn(

openapi_core/deserializing/parameters/factories.py

@@ -5,15 +5,9 @@
 from openapi_core.deserializing.parameters.datatypes import (
     DeserializerCallable,
 )
-from openapi_core.deserializing.parameters.deserializers import (
-    BaseParameterDeserializer,
-)
 from openapi_core.deserializing.parameters.deserializers import (
     CallableParameterDeserializer,
 )
-from openapi_core.deserializing.parameters.deserializers import (
-    UnsupportedStyleDeserializer,
-)
 from openapi_core.deserializing.parameters.util import split
 from openapi_core.schema.parameters import get_style
 from openapi_core.spec import Spec
@@ -28,13 +22,10 @@ class ParameterDeserializersFactory:
         "deepObject": partial(re.split, pattern=r"\[|\]"),
     }
 
-    def create(self, param_or_header: Spec) -> BaseParameterDeserializer:
+    def create(self, param_or_header: Spec) -> CallableParameterDeserializer:
         style = get_style(param_or_header)
 
-        if style not in self.PARAMETER_STYLE_DESERIALIZERS:
-            return UnsupportedStyleDeserializer(param_or_header, style)
-
-        deserialize_callable = self.PARAMETER_STYLE_DESERIALIZERS[style]
+        deserialize_callable = self.PARAMETER_STYLE_DESERIALIZERS.get(style)
         return CallableParameterDeserializer(
             param_or_header, style, deserialize_callable
         )

openapi_core/unmarshalling/request/unmarshallers.py

@@ -6,6 +6,9 @@
 from openapi_core.deserializing.media_types import (
     media_type_deserializers_factory,
 )
+from openapi_core.deserializing.media_types.datatypes import (
+    MediaTypeDeserializersDict,
+)
 from openapi_core.deserializing.media_types.factories import (
     MediaTypeDeserializersFactory,
 )
@@ -90,6 +93,9 @@ def __init__(
         schema_validators_factory: Optional[SchemaValidatorsFactory] = None,
         format_validators: Optional[FormatValidatorsDict] = None,
         extra_format_validators: Optional[FormatValidatorsDict] = None,
+        extra_media_type_deserializers: Optional[
+            MediaTypeDeserializersDict
+        ] = None,
         security_provider_factory: SecurityProviderFactory = security_provider_factory,
         schema_unmarshallers_factory: Optional[
             SchemaUnmarshallersFactory
@@ -107,6 +113,7 @@ def __init__(
             schema_validators_factory=schema_validators_factory,
             format_validators=format_validators,
             extra_format_validators=extra_format_validators,
+            extra_media_type_deserializers=extra_media_type_deserializers,
             schema_unmarshallers_factory=schema_unmarshallers_factory,
             format_unmarshallers=format_unmarshallers,
             extra_format_unmarshallers=extra_format_unmarshallers,
@@ -121,6 +128,7 @@ def __init__(
             schema_validators_factory=schema_validators_factory,
             format_validators=format_validators,
             extra_format_validators=extra_format_validators,
+            extra_media_type_deserializers=extra_media_type_deserializers,
             security_provider_factory=security_provider_factory,
         )
 

openapi_core/unmarshalling/unmarshallers.py

@@ -8,6 +8,9 @@
 from openapi_core.deserializing.media_types import (
     media_type_deserializers_factory,
 )
+from openapi_core.deserializing.media_types.datatypes import (
+    MediaTypeDeserializersDict,
+)
 from openapi_core.deserializing.media_types.factories import (
     MediaTypeDeserializersFactory,
 )
@@ -42,6 +45,9 @@ def __init__(
         schema_validators_factory: Optional[SchemaValidatorsFactory] = None,
         format_validators: Optional[FormatValidatorsDict] = None,
         extra_format_validators: Optional[FormatValidatorsDict] = None,
+        extra_media_type_deserializers: Optional[
+            MediaTypeDeserializersDict
+        ] = None,
         schema_unmarshallers_factory: Optional[
             SchemaUnmarshallersFactory
         ] = None,
@@ -61,6 +67,7 @@ def __init__(
             schema_validators_factory=schema_validators_factory,
             format_validators=format_validators,
             extra_format_validators=extra_format_validators,
+            extra_media_type_deserializers=extra_media_type_deserializers,
         )
         self.schema_unmarshallers_factory = (
             schema_unmarshallers_factory or self.schema_unmarshallers_factory

openapi_core/validation/request/validators.py

@@ -12,6 +12,9 @@
 from openapi_core.deserializing.media_types import (
     media_type_deserializers_factory,
 )
+from openapi_core.deserializing.media_types.datatypes import (
+    MediaTypeDeserializersDict,
+)
 from openapi_core.deserializing.media_types.factories import (
     MediaTypeDeserializersFactory,
 )
@@ -68,6 +71,9 @@ def __init__(
         schema_validators_factory: Optional[SchemaValidatorsFactory] = None,
         format_validators: Optional[FormatValidatorsDict] = None,
         extra_format_validators: Optional[FormatValidatorsDict] = None,
+        extra_media_type_deserializers: Optional[
+            MediaTypeDeserializersDict
+        ] = None,
         security_provider_factory: SecurityProviderFactory = security_provider_factory,
     ):
         super().__init__(
@@ -79,6 +85,7 @@ def __init__(
             schema_validators_factory=schema_validators_factory,
             format_validators=format_validators,
             extra_format_validators=extra_format_validators,
+            extra_media_type_deserializers=extra_media_type_deserializers,
         )
         self.security_provider_factory = security_provider_factory