Unmarshallers format validators refactor

Author: p1c2u

openapi_core/unmarshalling/schemas/__init__.py

@@ -1,31 +1,60 @@
-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_schema_validator._format import oas30_format_checker
+from openapi_schema_validator._format import oas31_format_checker
 
 from openapi_core.unmarshalling.schemas.enums import ValidationContext
 from openapi_core.unmarshalling.schemas.factories import (
     SchemaUnmarshallersFactory,
 )
+from openapi_core.unmarshalling.schemas.formatters import Formatter
+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_number
+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,
+    base_format_checker=oas30_format_checker,
+    format_unmarshallers=oas30_format_unmarshallers,
     context=ValidationContext.REQUEST,
 )
 
 oas30_response_schema_unmarshallers_factory = SchemaUnmarshallersFactory(
-    OAS30Validator,
+    OAS30ReadValidator,
+    base_format_checker=oas30_format_checker,
+    format_unmarshallers=oas30_format_unmarshallers,
     context=ValidationContext.RESPONSE,
 )
 
 oas31_schema_unmarshallers_factory = SchemaUnmarshallersFactory(
     OAS31Validator,
+    base_format_checker=oas31_format_checker,
+    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,10 @@
+from typing import Any
+from typing import Callable
 from typing import Dict
 from typing import Optional
 
 from openapi_core.unmarshalling.schemas.formatters import Formatter
 
 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,31 @@ def __str__(self) -> str:
 
 
 @dataclass
-class InvalidSchemaFormatValue(UnmarshallerError):
-    """Value failed to format with formatter"""
+class InvalidFormatValue(UnmarshallerError):
+    """Value not valid for format"""
 
     value: str
     type: str
-    original_exception: Exception
 
     def __str__(self) -> str:
-        return (
-            "Failed to format value {value} to format {type}: {exception}"
-        ).format(
+        return ("value {value} not valid for format {type}").format(
             value=self.value,
             type=self.type,
-            exception=self.original_exception,
         )
 
 
-@dataclass
-class FormatterNotFoundError(UnmarshallerError):
-    """Formatter not found to unmarshal"""
+class FormatUnmarshalError(UnmarshallerError):
+    """Unable to unmarshal value for format"""
 
-    type_format: str
+    value: str
+    type: str
+    original_exception: Exception
 
     def __str__(self) -> str:
-        return f"Formatter not found for {self.type_format} format"
+        return (
+            "Unable to unmarshal value {value} for format {type}: {exception}"
+        ).format(
+            value=self.value,
+            type=self.type,
+            exception=self.original_exception,
+        )

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,16 +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 UnmarshallersDict
 from openapi_core.unmarshalling.schemas.enums import ValidationContext
-from openapi_core.unmarshalling.schemas.exceptions import (
-    FormatterNotFoundError,
-)
 from openapi_core.unmarshalling.schemas.formatters import Formatter
 from openapi_core.unmarshalling.schemas.unmarshallers import AnyUnmarshaller
 from openapi_core.unmarshalling.schemas.unmarshallers import ArrayUnmarshaller
@@ -47,39 +48,97 @@
 
 
 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()
+            for formatters_list in [self.formatters, self.custom_formatters]
+            for name, formatter in formatters_list.items()
         }
-        format_checker = build_format_checker(**custom_format_checks)
-        kwargs = {
-            "resolver": resolver,
-            "format_checker": format_checker,
-        }
-        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) -> Callable[[Any], bool]:
+        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) -> Callable[[Any], bool]:
+        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 SchemaFormattersFactory:
+    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[Formatter]:
+        if schema_format in self.custom_formatters:
+            return self.custom_formatters[schema_format]
+        if schema_format in self.formatters:
+            return self.formatters[schema_format]
+        if schema_format in self.format_unmarshallers:
+            validate_callable = self.validators_factory.get_checker(
+                schema_format
+            )
+            format_callable = self.format_unmarshallers[schema_format]
+            return Formatter.from_callables(validate_callable, format_callable)
+
+        return None
 
 
 class SchemaUnmarshallersFactory:
@@ -104,21 +163,40 @@ 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 = {}
+        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 formatters_factory(self) -> SchemaFormattersFactory:
+        return SchemaFormattersFactory(
+            self.validators_factory,
+            self.formatters,
+            self.format_unmarshallers,
             self.custom_formatters,
-            self.context,
         )
 
     def create(
@@ -134,7 +212,7 @@ def create(
         validator = self.validators_factory.create(schema)
 
         schema_format = schema.getkey("format")
-        formatter = self.custom_formatters.get(schema_format)
+        formatter = self.formatters_factory.create(schema_format)
 
         schema_type = type_override or schema.getkey("type", "any")
         if isinstance(schema_type, Iterable) and not isinstance(