Format validators and Format unmarshallers; deprecate custom formatters

Author: p1c2u

docs/customizations.rst

@@ -42,42 +42,54 @@ Pass custom defined media type deserializers dictionary with supported mimetypes
        media_type_deserializers_factory=media_type_deserializers_factory,
    )
 
-Formats
--------
+Format validators
+-----------------
 
 OpenAPI defines a ``format`` keyword that hints at how a value should be interpreted, e.g. a ``string`` with the type ``date`` should conform to the RFC 3339 date format.
 
-Openapi-core comes with a set of built-in formatters, but it's also possible to add custom formatters in `SchemaUnmarshallersFactory` and pass it to `RequestValidator` or `ResponseValidator`.
+OpenAPI comes with a set of built-in format validators, but it's also possible to add custom ones.
 
 Here's how you could add support for a ``usdate`` format that handles dates of the form MM/DD/YYYY:
 
 .. code-block:: python
 
-   from openapi_core.unmarshalling.schemas.factories import SchemaUnmarshallersFactory
-   from openapi_schema_validator import OAS30Validator
-   from datetime import datetime
    import re
 
-   class USDateFormatter:
-       def validate(self, value) -> bool:
-           return bool(re.match(r"^\d{1,2}/\d{1,2}/\d{4}$", value))
-
-       def format(self, value):
-           return datetime.strptime(value, "%m/%d/%y").date
-
+   def validate_usdate(value):
+       return bool(re.match(r"^\d{1,2}/\d{1,2}/\d{4}$", value))
 
-   custom_formatters = {
-       'usdate': USDateFormatter(),
+   extra_format_validators = {
+       'usdate': validate_usdate,
    }
-   schema_unmarshallers_factory = SchemaUnmarshallersFactory(
-       OAS30Validator,
-       custom_formatters=custom_formatters,
-   )
 
    result = validate_response(
        request, response,
        spec=spec,
-       cls=ResponseValidator,
-       schema_unmarshallers_factory=schema_unmarshallers_factory,
+       extra_format_validators=extra_format_validators,
    )
 
+Format unmarshallers
+--------------------
+
+Based on ``format`` keyword, openapi-core can also unmarshal values to specific formats.
+
+Openapi-core comes with a set of built-in format unmarshallers, but it's also possible to add custom ones.
+
+Here's an example with the ``usdate`` format that converts a value to date object:
+
+.. code-block:: python
+
+   from datetime import datetime
+
+   def unmarshal_usdate(value):
+       return datetime.strptime(value, "%m/%d/%y").date
+
+   extra_format_unmarshallers = {
+       'usdate': unmarshal_usdate,
+   }
+
+   result = unmarshal_response(
+       request, response,
+       spec=spec,
+       extra_format_unmarshallers=extra_format_unmarshallers,
+   )

openapi_core/unmarshalling/request/unmarshallers.py

@@ -32,6 +32,9 @@
 from openapi_core.unmarshalling.schemas import (
     oas31_schema_unmarshallers_factory,
 )
+from openapi_core.unmarshalling.schemas.datatypes import (
+    FormatUnmarshallersDict,
+)
 from openapi_core.unmarshalling.schemas.factories import (
     SchemaUnmarshallersFactory,
 )
@@ -72,6 +75,7 @@
     V31WebhookRequestValidator,
 )
 from openapi_core.validation.request.validators import WebhookRequestValidator
+from openapi_core.validation.schemas.datatypes import FormatValidatorsDict
 from openapi_core.validation.schemas.factories import SchemaValidatorsFactory
 
 
@@ -84,10 +88,14 @@ def __init__(
         parameter_deserializers_factory: ParameterDeserializersFactory = parameter_deserializers_factory,
         media_type_deserializers_factory: MediaTypeDeserializersFactory = media_type_deserializers_factory,
         schema_validators_factory: Optional[SchemaValidatorsFactory] = None,
+        format_validators: Optional[FormatValidatorsDict] = None,
+        extra_format_validators: Optional[FormatValidatorsDict] = None,
         security_provider_factory: SecurityProviderFactory = security_provider_factory,
         schema_unmarshallers_factory: Optional[
             SchemaUnmarshallersFactory
         ] = None,
+        format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
+        extra_format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
     ):
         BaseUnmarshaller.__init__(
             self,
@@ -97,7 +105,11 @@ def __init__(
             parameter_deserializers_factory=parameter_deserializers_factory,
             media_type_deserializers_factory=media_type_deserializers_factory,
             schema_validators_factory=schema_validators_factory,
+            format_validators=format_validators,
+            extra_format_validators=extra_format_validators,
             schema_unmarshallers_factory=schema_unmarshallers_factory,
+            format_unmarshallers=format_unmarshallers,
+            extra_format_unmarshallers=extra_format_unmarshallers,
         )
         BaseRequestValidator.__init__(
             self,
@@ -107,6 +119,8 @@ def __init__(
             parameter_deserializers_factory=parameter_deserializers_factory,
             media_type_deserializers_factory=media_type_deserializers_factory,
             schema_validators_factory=schema_validators_factory,
+            format_validators=format_validators,
+            extra_format_validators=extra_format_validators,
             security_provider_factory=security_provider_factory,
         )
 

openapi_core/unmarshalling/schemas/datatypes.py

@@ -1,12 +1,6 @@
 from typing import Any
 from typing import Callable
 from typing import Dict
-from typing import Optional
-
-from openapi_core.unmarshalling.schemas.formatters import Formatter
 
 FormatUnmarshaller = Callable[[Any], Any]
-
-CustomFormattersDict = Dict[str, Formatter]
-FormattersDict = Dict[Optional[str], Formatter]
-UnmarshallersDict = Dict[str, Callable[[Any], Any]]
+FormatUnmarshallersDict = Dict[str, FormatUnmarshaller]

openapi_core/unmarshalling/schemas/factories.py

@@ -2,15 +2,10 @@
 import warnings
 from typing import Optional
 
-if sys.version_info >= (3, 8):
-    from functools import cached_property
-else:
-    from backports.cached_property import cached_property
-
 from openapi_core.spec import Spec
-from openapi_core.unmarshalling.schemas.datatypes import CustomFormattersDict
-from openapi_core.unmarshalling.schemas.datatypes import FormatUnmarshaller
-from openapi_core.unmarshalling.schemas.datatypes import UnmarshallersDict
+from openapi_core.unmarshalling.schemas.datatypes import (
+    FormatUnmarshallersDict,
+)
 from openapi_core.unmarshalling.schemas.exceptions import (
     FormatterNotFoundError,
 )
@@ -19,6 +14,8 @@
 )
 from openapi_core.unmarshalling.schemas.unmarshallers import SchemaUnmarshaller
 from openapi_core.unmarshalling.schemas.unmarshallers import TypesUnmarshaller
+from openapi_core.validation.schemas.datatypes import CustomFormattersDict
+from openapi_core.validation.schemas.datatypes import FormatValidatorsDict
 from openapi_core.validation.schemas.factories import SchemaValidatorsFactory
 
 
@@ -27,55 +24,74 @@ def __init__(
         self,
         schema_validators_factory: SchemaValidatorsFactory,
         types_unmarshaller: TypesUnmarshaller,
-        format_unmarshallers: Optional[UnmarshallersDict] = None,
+        format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
         custom_formatters: Optional[CustomFormattersDict] = None,
     ):
         self.schema_validators_factory = schema_validators_factory
         self.types_unmarshaller = types_unmarshaller
-        if custom_formatters is None:
-            custom_formatters = {}
         if format_unmarshallers is None:
             format_unmarshallers = {}
         self.format_unmarshallers = format_unmarshallers
+        if custom_formatters is None:
+            custom_formatters = {}
+        else:
+            warnings.warn(
+                "custom_formatters is deprecated. "
+                "Use extra_format_validators to validate custom formats "
+                "and use extra_format_unmarshallers to unmarshal custom formats.",
+                DeprecationWarning,
+            )
         self.custom_formatters = custom_formatters
 
-    @cached_property
-    def formats_unmarshaller(self) -> FormatsUnmarshaller:
-        return FormatsUnmarshaller(
-            self.format_unmarshallers,
-            self.custom_formatters,
-        )
-
-    def create(self, schema: Spec) -> SchemaUnmarshaller:
+    def create(
+        self,
+        schema: Spec,
+        format_validators: Optional[FormatValidatorsDict] = None,
+        format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
+        extra_format_validators: Optional[FormatValidatorsDict] = None,
+        extra_format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
+    ) -> SchemaUnmarshaller:
         """Create unmarshaller from the schema."""
         if schema is None:
             raise TypeError("Invalid schema")
 
         if schema.getkey("deprecated", False):
             warnings.warn("The schema is deprecated", DeprecationWarning)
 
-        formatters_checks = {
-            name: formatter.validate
-            for name, formatter in self.custom_formatters.items()
-        }
+        if extra_format_validators is None:
+            extra_format_validators = {}
+        extra_format_validators.update(
+            {
+                name: formatter.validate
+                for name, formatter in self.custom_formatters.items()
+            }
+        )
         schema_validator = self.schema_validators_factory.create(
-            schema, **formatters_checks
+            schema,
+            format_validators=format_validators,
+            extra_format_validators=extra_format_validators,
         )
 
         schema_format = schema.getkey("format")
 
+        formats_unmarshaller = FormatsUnmarshaller(
+            format_unmarshallers or self.format_unmarshallers,
+            extra_format_unmarshallers,
+            self.custom_formatters,
+        )
+
         # FIXME: don;t raise exception on unknown format
+        # See https://github.com/p1c2u/openapi-core/issues/515
         if (
             schema_format
-            and schema_format
-            not in self.schema_validators_factory.format_checker.checkers
-            and schema_format not in self.custom_formatters
+            and schema_format not in schema_validator
+            and schema_format not in formats_unmarshaller
         ):
             raise FormatterNotFoundError(schema_format)
 
         return SchemaUnmarshaller(
             schema,
             schema_validator,
             self.types_unmarshaller,
-            self.formats_unmarshaller,
+            formats_unmarshaller,
         )

openapi_core/unmarshalling/schemas/unmarshallers.py

@@ -12,11 +12,13 @@
 from openapi_core.extensions.models.factories import ModelPathFactory
 from openapi_core.schema.schemas import get_properties
 from openapi_core.spec import Spec
-from openapi_core.unmarshalling.schemas.datatypes import CustomFormattersDict
 from openapi_core.unmarshalling.schemas.datatypes import FormatUnmarshaller
-from openapi_core.unmarshalling.schemas.datatypes import UnmarshallersDict
+from openapi_core.unmarshalling.schemas.datatypes import (
+    FormatUnmarshallersDict,
+)
 from openapi_core.unmarshalling.schemas.exceptions import FormatUnmarshalError
 from openapi_core.unmarshalling.schemas.exceptions import UnmarshallerError
+from openapi_core.validation.schemas.datatypes import CustomFormattersDict
 from openapi_core.validation.schemas.validators import SchemaValidator
 
 log = logging.getLogger(__name__)
@@ -212,12 +214,16 @@ def get_unmarshaller(
 class FormatsUnmarshaller:
     def __init__(
         self,
-        format_unmarshallers: Optional[UnmarshallersDict] = None,
+        format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
+        extra_format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
         custom_formatters: Optional[CustomFormattersDict] = None,
     ):
         if format_unmarshallers is None:
             format_unmarshallers = {}
         self.format_unmarshallers = format_unmarshallers
+        if extra_format_unmarshallers is None:
+            extra_format_unmarshallers = {}
+        self.extra_format_unmarshallers = extra_format_unmarshallers
         if custom_formatters is None:
             custom_formatters = {}
         self.custom_formatters = custom_formatters
@@ -237,11 +243,24 @@ def get_unmarshaller(
         if schema_format in self.custom_formatters:
             formatter = self.custom_formatters[schema_format]
             return formatter.format
+        if schema_format in self.extra_format_unmarshallers:
+            return self.extra_format_unmarshallers[schema_format]
         if schema_format in self.format_unmarshallers:
             return self.format_unmarshallers[schema_format]
 
         return None
 
+    def __contains__(self, schema_format: str) -> bool:
+        format_unmarshallers_dicts: List[Mapping[str, Any]] = [
+            self.custom_formatters,
+            self.extra_format_unmarshallers,
+            self.format_unmarshallers,
+        ]
+        for content in format_unmarshallers_dicts:
+            if schema_format in content:
+                return True
+        return False
+
 
 class SchemaUnmarshaller:
     def __init__(

openapi_core/unmarshalling/unmarshallers.py

@@ -18,9 +18,13 @@
     ParameterDeserializersFactory,
 )
 from openapi_core.spec import Spec
+from openapi_core.unmarshalling.schemas.datatypes import (
+    FormatUnmarshallersDict,
+)
 from openapi_core.unmarshalling.schemas.factories import (
     SchemaUnmarshallersFactory,
 )
+from openapi_core.validation.schemas.datatypes import FormatValidatorsDict
 from openapi_core.validation.schemas.factories import SchemaValidatorsFactory
 from openapi_core.validation.validators import BaseValidator
 
@@ -36,9 +40,13 @@ def __init__(
         parameter_deserializers_factory: ParameterDeserializersFactory = parameter_deserializers_factory,
         media_type_deserializers_factory: MediaTypeDeserializersFactory = media_type_deserializers_factory,
         schema_validators_factory: Optional[SchemaValidatorsFactory] = None,
+        format_validators: Optional[FormatValidatorsDict] = None,
+        extra_format_validators: Optional[FormatValidatorsDict] = None,
         schema_unmarshallers_factory: Optional[
             SchemaUnmarshallersFactory
         ] = None,
+        format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
+        extra_format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
     ):
         if schema_validators_factory is None and schema_unmarshallers_factory:
             schema_validators_factory = (
@@ -51,6 +59,8 @@ def __init__(
             parameter_deserializers_factory=parameter_deserializers_factory,
             media_type_deserializers_factory=media_type_deserializers_factory,
             schema_validators_factory=schema_validators_factory,
+            format_validators=format_validators,
+            extra_format_validators=extra_format_validators,
         )
         self.schema_unmarshallers_factory = (
             schema_unmarshallers_factory or self.schema_unmarshallers_factory
@@ -59,9 +69,17 @@ def __init__(
             raise NotImplementedError(
                 "schema_unmarshallers_factory is not assigned"
             )
+        self.format_unmarshallers = format_unmarshallers
+        self.extra_format_unmarshallers = extra_format_unmarshallers
 
     def _unmarshal_schema(self, schema: Spec, value: Any) -> Any:
-        unmarshaller = self.schema_unmarshallers_factory.create(schema)
+        unmarshaller = self.schema_unmarshallers_factory.create(
+            schema,
+            format_validators=self.format_validators,
+            extra_format_validators=self.extra_format_validators,
+            format_unmarshallers=self.format_unmarshallers,
+            extra_format_unmarshallers=self.extra_format_unmarshallers,
+        )
         return unmarshaller.unmarshal(value)
 
     def _get_param_or_header_value(