Unmarshallers format validators refactor

Author: p1c2u

openapi_core/unmarshalling/schemas/__init__.py

@@ -1,31 +1,53 @@
-from openapi_schema_validator import OAS30Validator
+from functools import partial
+
+from isodate.isodatetime import parse_datetime
+from openapi_schema_validator import OAS30ReadValidator
+from openapi_schema_validator import OAS30WriteValidator
 from openapi_schema_validator import OAS31Validator
 
 from openapi_core.unmarshalling.schemas.enums import ValidationContext
 from openapi_core.unmarshalling.schemas.factories import (
     SchemaUnmarshallersFactory,
 )
+from openapi_core.unmarshalling.schemas.util import format_byte
+from openapi_core.unmarshalling.schemas.util import format_date
+from openapi_core.unmarshalling.schemas.util import format_uuid
 
 __all__ = [
+    "oas30_format_unmarshallers",
+    "oas31_format_unmarshallers",
     "oas30_request_schema_unmarshallers_factory",
     "oas30_response_schema_unmarshallers_factory",
     "oas31_request_schema_unmarshallers_factory",
     "oas31_response_schema_unmarshallers_factory",
     "oas31_schema_unmarshallers_factory",
 ]
 
+oas30_format_unmarshallers = {
+    # string compatible
+    "date": format_date,
+    "date-time": parse_datetime,
+    "binary": bytes,
+    "uuid": format_uuid,
+    "byte": format_byte,
+}
+oas31_format_unmarshallers = oas30_format_unmarshallers
+
 oas30_request_schema_unmarshallers_factory = SchemaUnmarshallersFactory(
-    OAS30Validator,
+    OAS30WriteValidator,
+    format_unmarshallers=oas30_format_unmarshallers,
     context=ValidationContext.REQUEST,
 )
 
 oas30_response_schema_unmarshallers_factory = SchemaUnmarshallersFactory(
-    OAS30Validator,
+    OAS30ReadValidator,
+    format_unmarshallers=oas30_format_unmarshallers,
     context=ValidationContext.RESPONSE,
 )
 
 oas31_schema_unmarshallers_factory = SchemaUnmarshallersFactory(
     OAS31Validator,
+    format_unmarshallers=oas31_format_unmarshallers,
 )
 
 # alias to v31 version (request/response are the same bcs no context needed)

openapi_core/unmarshalling/schemas/datatypes.py

@@ -1,7 +1,13 @@
+from typing import Any
+from typing import Callable
 from typing import Dict
 from typing import Optional
 
 from openapi_core.unmarshalling.schemas.formatters import Formatter
 
+FormatValidator = Callable[[Any], bool]
+FormatUnmarshaller = Callable[[Any], Any]
+
 CustomFormattersDict = Dict[str, Formatter]
 FormattersDict = Dict[Optional[str], Formatter]
+UnmarshallersDict = Dict[str, Callable[[Any], Any]]

openapi_core/unmarshalling/schemas/exceptions.py

@@ -19,6 +19,8 @@ class UnmarshallerError(UnmarshalError):
 
 @dataclass
 class InvalidSchemaValue(ValidateError):
+    """Value not valid for schema"""
+
     value: str
     type: str
     schema_errors: Iterable[Exception] = field(default_factory=list)
@@ -30,28 +32,27 @@ def __str__(self) -> str:
 
 
 @dataclass
-class InvalidSchemaFormatValue(UnmarshallerError):
-    """Value failed to format with formatter"""
+class FormatterNotFoundError(UnmarshallerError):
+    """Formatter not found to unmarshal"""
+
+    type_format: str
+
+    def __str__(self) -> str:
+        return f"Formatter not found for {self.type_format} format"
+
+
+class FormatUnmarshalError(UnmarshallerError):
+    """Unable to unmarshal value for format"""
 
     value: str
     type: str
     original_exception: Exception
 
     def __str__(self) -> str:
         return (
-            "Failed to format value {value} to format {type}: {exception}"
+            "Unable to unmarshal value {value} for format {type}: {exception}"
         ).format(
             value=self.value,
             type=self.type,
             exception=self.original_exception,
         )
-
-
-@dataclass
-class FormatterNotFoundError(UnmarshallerError):
-    """Formatter not found to unmarshal"""
-
-    type_format: str
-
-    def __str__(self) -> str:
-        return f"Formatter not found for {self.type_format} format"

openapi_core/unmarshalling/schemas/factories.py

@@ -1,6 +1,8 @@
 import sys
 import warnings
+from functools import partial
 from typing import Any
+from typing import Callable
 from typing import Dict
 from typing import Iterable
 from typing import Optional
@@ -11,12 +13,15 @@
     from functools import cached_property
 else:
     from backports.cached_property import cached_property
+from jsonschema._format import FormatChecker
 from jsonschema.protocols import Validator
 from openapi_schema_validator import OAS30Validator
 
 from openapi_core.spec import Spec
 from openapi_core.unmarshalling.schemas.datatypes import CustomFormattersDict
-from openapi_core.unmarshalling.schemas.datatypes import FormattersDict
+from openapi_core.unmarshalling.schemas.datatypes import FormatUnmarshaller
+from openapi_core.unmarshalling.schemas.datatypes import FormatValidator
+from openapi_core.unmarshalling.schemas.datatypes import UnmarshallersDict
 from openapi_core.unmarshalling.schemas.enums import ValidationContext
 from openapi_core.unmarshalling.schemas.exceptions import (
     FormatterNotFoundError,
@@ -47,38 +52,95 @@
 
 
 class SchemaValidatorsFactory:
-    CONTEXTS = {
-        ValidationContext.REQUEST: "write",
-        ValidationContext.RESPONSE: "read",
-    }
-
     def __init__(
         self,
         schema_validator_class: Type[Validator],
+        base_format_checker: Optional[FormatChecker] = None,
+        formatters: Optional[CustomFormattersDict] = None,
+        format_unmarshallers: Optional[UnmarshallersDict] = None,
         custom_formatters: Optional[CustomFormattersDict] = None,
-        context: Optional[ValidationContext] = None,
     ):
         self.schema_validator_class = schema_validator_class
+        if base_format_checker is None:
+            base_format_checker = self.schema_validator_class.FORMAT_CHECKER
+        self.base_format_checker = base_format_checker
+        if formatters is None:
+            formatters = {}
+        self.formatters = formatters
+        if format_unmarshallers is None:
+            format_unmarshallers = {}
+        self.format_unmarshallers = format_unmarshallers
         if custom_formatters is None:
             custom_formatters = {}
         self.custom_formatters = custom_formatters
-        self.context = context
 
-    def create(self, schema: Spec) -> Validator:
-        resolver = schema.accessor.resolver  # type: ignore
-        custom_format_checks = {
+    @cached_property
+    def format_checker(self) -> FormatChecker:
+        format_checks: Dict[str, Callable[[Any], bool]] = {
             name: formatter.validate
-            for name, formatter in self.custom_formatters.items()
-        }
-        format_checker = build_format_checker(**custom_format_checks)
-        kwargs = {
-            "resolver": resolver,
-            "format_checker": format_checker,
+            for formatters_list in [self.formatters, self.custom_formatters]
+            for name, formatter in formatters_list.items()
         }
-        if self.context is not None:
-            kwargs[self.CONTEXTS[self.context]] = True
+        format_checks.update(
+            {
+                name: self._create_checker(name)
+                for name, _ in self.format_unmarshallers.items()
+            }
+        )
+        return build_format_checker(self.base_format_checker, **format_checks)
+
+    def _create_checker(self, name: str) -> FormatValidator:
+        if name in self.base_format_checker.checkers:
+            return partial(self.base_format_checker.check, format=name)
+
+        return lambda x: True
+
+    def get_checker(self, name: str) -> FormatValidator:
+        if name in self.format_checker.checkers:
+            return partial(self.format_checker.check, format=name)
+
+        return lambda x: True
+
+    def create(self, schema: Spec) -> Validator:
+        resolver = schema.accessor.resolver  # type: ignore
         with schema.open() as schema_dict:
-            return self.schema_validator_class(schema_dict, **kwargs)
+            return self.schema_validator_class(
+                schema_dict,
+                resolver=resolver,
+                format_checker=self.format_checker,
+            )
+
+
+class SchemaFormatUnmarshallersFactory:
+    def __init__(
+        self,
+        validators_factory: SchemaValidatorsFactory,
+        formatters: Optional[CustomFormattersDict] = None,
+        format_unmarshallers: Optional[UnmarshallersDict] = None,
+        custom_formatters: Optional[CustomFormattersDict] = None,
+    ):
+        self.validators_factory = validators_factory
+        if formatters is None:
+            formatters = {}
+        self.formatters = formatters
+        if format_unmarshallers is None:
+            format_unmarshallers = {}
+        self.format_unmarshallers = format_unmarshallers
+        if custom_formatters is None:
+            custom_formatters = {}
+        self.custom_formatters = custom_formatters
+
+    def create(self, schema_format: str) -> Optional[FormatUnmarshaller]:
+        if schema_format in self.custom_formatters:
+            formatter = self.custom_formatters[schema_format]
+            return formatter.format
+        if schema_format in self.formatters:
+            formatter = self.formatters[schema_format]
+            return formatter.format
+        if schema_format in self.format_unmarshallers:
+            return self.format_unmarshallers[schema_format]
+
+        return None
 
 
 class SchemaUnmarshallersFactory:
@@ -102,23 +164,59 @@ class SchemaUnmarshallersFactory:
     def __init__(
         self,
         schema_validator_class: Type[Validator],
+        base_format_checker: Optional[FormatChecker] = None,
+        formatters: Optional[CustomFormattersDict] = None,
+        format_unmarshallers: Optional[UnmarshallersDict] = None,
         custom_formatters: Optional[CustomFormattersDict] = None,
         context: Optional[ValidationContext] = None,
     ):
         self.schema_validator_class = schema_validator_class
+        self.base_format_checker = base_format_checker
         if custom_formatters is None:
             custom_formatters = {}
+        else:
+            warnings.warn(
+                "custom_formatters is deprecated. "
+                "Register new checks to FormatChecker to validate custom formats "
+                "and add format_unmarshallers to unmarshal custom formats.",
+                DeprecationWarning,
+            )
+        self.formatters = formatters
+        if format_unmarshallers is None:
+            format_unmarshallers = {}
+        self.format_unmarshallers = format_unmarshallers
         self.custom_formatters = custom_formatters
         self.context = context
 
     @cached_property
     def validators_factory(self) -> SchemaValidatorsFactory:
         return SchemaValidatorsFactory(
             self.schema_validator_class,
+            self.base_format_checker,
+            self.formatters,
+            self.format_unmarshallers,
+            self.custom_formatters,
+        )
+
+    @cached_property
+    def format_unmarshallers_factory(self) -> SchemaFormatUnmarshallersFactory:
+        return SchemaFormatUnmarshallersFactory(
+            self.validators_factory,
+            self.formatters,
+            self.format_unmarshallers,
             self.custom_formatters,
-            self.context,
         )
 
+    def get_format_validator(
+        self, validator: Validator, schema_format: str
+    ) -> Optional[FormatValidator]:
+        if schema_format in validator.format_checker.checkers:
+            return partial(
+                validator.format_checker.check, format=schema_format
+            )
+
+        return None
+
     def create(
         self, schema: Spec, type_override: Optional[str] = None
     ) -> BaseSchemaUnmarshaller:
@@ -132,7 +230,19 @@ def create(
         validator = self.validators_factory.create(schema)
 
         schema_format = schema.getkey("format")
-        formatter = self.custom_formatters.get(schema_format)
+
+        # FIXME: don;t raise exception on unknown format
+        if (
+            schema_format
+            and schema_format
+            not in self.validators_factory.format_checker.checkers
+        ):
+            raise FormatterNotFoundError(schema_format)
+
+        format_unmarshaller = self.format_unmarshallers_factory.create(
+            schema_format
+        )
+        format_validator = self.get_format_validator(validator, schema_format)
 
         schema_type = type_override or schema.getkey("type", "any")
         if isinstance(schema_type, Iterable) and not isinstance(
@@ -141,7 +251,7 @@ def create(
             return MultiTypeUnmarshaller(
                 schema,
                 validator,
-                formatter,
+                format_unmarshaller,
                 self.validators_factory,
                 self,
                 context=self.context,
@@ -151,7 +261,7 @@ def create(
             return complex_klass(
                 schema,
                 validator,
-                formatter,
+                format_unmarshaller,
                 self.validators_factory,
                 self,
                 context=self.context,
@@ -161,7 +271,7 @@ def create(
         return klass(
             schema,
             validator,
-            formatter,
+            format_unmarshaller,
             self.validators_factory,
             self,
         )