Object caster

Author: p1c2u

openapi_core/casting/schemas/__init__.py

@@ -1,5 +1,65 @@
+from collections import OrderedDict
+
+from openapi_core.casting.schemas.casters import ArrayCaster
+from openapi_core.casting.schemas.casters import BooleanCaster
+from openapi_core.casting.schemas.casters import DummyCaster
+from openapi_core.casting.schemas.casters import IntegerCaster
+from openapi_core.casting.schemas.casters import NumberCaster
+from openapi_core.casting.schemas.casters import ObjectCaster
+from openapi_core.casting.schemas.casters import TypesCaster
 from openapi_core.casting.schemas.factories import SchemaCastersFactory
+from openapi_core.validation.schemas import (
+    oas30_read_schema_validators_factory,
+)
+from openapi_core.validation.schemas import (
+    oas30_write_schema_validators_factory,
+)
+from openapi_core.validation.schemas import oas31_schema_validators_factory
+
+__all__ = [
+    "oas30_write_schema_casters_factory",
+    "oas30_read_schema_casters_factory",
+    "oas31_schema_casters_factory",
+]
+
+oas30_casters_dict = OrderedDict(
+    [
+        ("object", ObjectCaster),
+        ("array", ArrayCaster),
+        ("boolean", BooleanCaster),
+        ("integer", IntegerCaster),
+        ("number", NumberCaster),
+        ("string", DummyCaster),
+    ]
+)
+oas31_casters_dict = oas30_casters_dict.copy()
+oas31_casters_dict.update(
+    {
+        "null": DummyCaster,
+    }
+)
+
+oas30_types_caster = TypesCaster(
+    oas30_casters_dict,
+    DummyCaster,
+)
+oas31_types_caster = TypesCaster(
+    oas31_casters_dict,
+    DummyCaster,
+    multi=DummyCaster,
+)
+
+oas30_write_schema_casters_factory = SchemaCastersFactory(
+    oas30_write_schema_validators_factory,
+    oas30_types_caster,
+)
 
-__all__ = ["schema_casters_factory"]
+oas30_read_schema_casters_factory = SchemaCastersFactory(
+    oas30_read_schema_validators_factory,
+    oas30_types_caster,
+)
 
-schema_casters_factory = SchemaCastersFactory()
+oas31_schema_casters_factory = SchemaCastersFactory(
+    oas31_schema_validators_factory,
+    oas31_types_caster,
+)

openapi_core/casting/schemas/casters.py

@@ -1,20 +1,36 @@
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
+from typing import Generic
+from typing import Iterable
 from typing import List
+from typing import Mapping
+from typing import Optional
+from typing import Type
+from typing import TypeVar
+from typing import Union
 
 from jsonschema_path import SchemaPath
 
 from openapi_core.casting.schemas.datatypes import CasterCallable
 from openapi_core.casting.schemas.exceptions import CastError
+from openapi_core.schema.schemas import get_properties
+from openapi_core.validation.schemas.validators import SchemaValidator
 
 if TYPE_CHECKING:
     from openapi_core.casting.schemas.factories import SchemaCastersFactory
 
 
-class BaseSchemaCaster:
-    def __init__(self, schema: SchemaPath):
+class PrimitiveCaster:
+    def __init__(
+        self,
+        schema: SchemaPath,
+        schema_validator: SchemaValidator,
+        schema_caster: "SchemaCaster",
+    ):
         self.schema = schema
+        self.schema_validator = schema_validator
+        self.schema_caster = schema_caster
 
     def __call__(self, value: Any) -> Any:
         if value is None:
@@ -26,42 +42,206 @@ def cast(self, value: Any) -> Any:
         raise NotImplementedError
 
 
-class CallableSchemaCaster(BaseSchemaCaster):
-    def __init__(self, schema: SchemaPath, caster_callable: CasterCallable):
-        super().__init__(schema)
-        self.caster_callable = caster_callable
+class DummyCaster(PrimitiveCaster):
+    def cast(self, value: Any) -> Any:
+        return value
+
+
+PrimitiveType = TypeVar("PrimitiveType")
+
+
+class PrimitiveTypeCaster(Generic[PrimitiveType], PrimitiveCaster):
+    primitive_type: Type[PrimitiveType] = NotImplemented
 
     def cast(self, value: Any) -> Any:
+        if isinstance(value, self.primitive_type):
+            return value
+
+        if not isinstance(value, (str, bytes)):
+            raise CastError(value, self.schema["type"])
+
         try:
-            return self.caster_callable(value)
+            return self.primitive_type(value)  # type: ignore [call-arg]
         except (ValueError, TypeError):
             raise CastError(value, self.schema["type"])
 
 
-class DummyCaster(BaseSchemaCaster):
-    def cast(self, value: Any) -> Any:
-        return value
+class IntegerCaster(PrimitiveTypeCaster[int]):
+    primitive_type = int
 
 
-class ComplexCaster(BaseSchemaCaster):
-    def __init__(
-        self, schema: SchemaPath, casters_factory: "SchemaCastersFactory"
-    ):
-        super().__init__(schema)
-        self.casters_factory = casters_factory
+class NumberCaster(PrimitiveTypeCaster[float]):
+    primitive_type = float
+
 
+class BooleanCaster(PrimitiveTypeCaster[bool]):
+    primitive_type = bool
 
-class ArrayCaster(ComplexCaster):
+
+class ArrayCaster(PrimitiveCaster):
     @property
-    def items_caster(self) -> BaseSchemaCaster:
-        return self.casters_factory.create(self.schema / "items")
+    def items_caster(self) -> "SchemaCaster":
+        # sometimes we don't have any schema i.e. free-form objects
+        items_schema = self.schema.get("items", SchemaPath.from_dict({}))
+        return self.schema_caster.evolve(items_schema)
 
     def cast(self, value: Any) -> List[Any]:
         # str and bytes are not arrays according to the OpenAPI spec
-        if isinstance(value, (str, bytes)):
+        if isinstance(value, (str, bytes)) or not isinstance(value, Iterable):
             raise CastError(value, self.schema["type"])
 
         try:
-            return list(map(self.items_caster, value))
+            return list(map(self.items_caster.cast, value))
         except (ValueError, TypeError):
             raise CastError(value, self.schema["type"])
+
+
+class ObjectCaster(PrimitiveCaster):
+    def cast(self, value: Any) -> Any:
+        return self._cast_proparties(value)
+
+    def evolve(self, schema: SchemaPath) -> "ObjectCaster":
+        cls = self.__class__
+
+        return cls(
+            schema,
+            self.schema_validator.evolve(schema),
+            self.schema_caster.evolve(schema),
+        )
+
+    def _cast_proparties(self, value: Any, schema_only: bool = False) -> Any:
+        if not isinstance(value, dict):
+            raise CastError(value, self.schema["type"])
+
+        one_of_schema = self.schema_validator.get_one_of_schema(value)
+        if one_of_schema is not None:
+            one_of_properties = self.evolve(one_of_schema)._cast_proparties(
+                value, schema_only=True
+            )
+            value.update(one_of_properties)
+
+        any_of_schemas = self.schema_validator.iter_any_of_schemas(value)
+        for any_of_schema in any_of_schemas:
+            any_of_properties = self.evolve(any_of_schema)._cast_proparties(
+                value, schema_only=True
+            )
+            value.update(any_of_properties)
+
+        all_of_schemas = self.schema_validator.iter_all_of_schemas(value)
+        for all_of_schema in all_of_schemas:
+            all_of_properties = self.evolve(all_of_schema)._cast_proparties(
+                value, schema_only=True
+            )
+            value.update(all_of_properties)
+
+        for prop_name, prop_schema in get_properties(self.schema).items():
+            try:
+                prop_value = value[prop_name]
+            except KeyError:
+                if "default" not in prop_schema:
+                    continue
+                prop_value = prop_schema["default"]
+
+            value[prop_name] = self.schema_caster.evolve(prop_schema).cast(
+                prop_value
+            )
+
+        if schema_only:
+            return value
+
+        additional_properties = self.schema.getkey(
+            "additionalProperties", True
+        )
+        if additional_properties is not False:
+            # free-form object
+            if additional_properties is True:
+                additional_prop_schema = SchemaPath.from_dict(
+                    {"nullable": True}
+                )
+            # defined schema
+            else:
+                additional_prop_schema = self.schema / "additionalProperties"
+            additional_prop_caster = self.schema_caster.evolve(
+                additional_prop_schema
+            )
+            for prop_name, prop_value in value.items():
+                if prop_name in value:
+                    continue
+                value[prop_name] = additional_prop_caster.cast(prop_value)
+
+        return value
+
+
+class TypesCaster:
+    casters: Mapping[str, Type[PrimitiveCaster]] = {}
+    multi: Optional[Type[PrimitiveCaster]] = None
+
+    def __init__(
+        self,
+        casters: Mapping[str, Type[PrimitiveCaster]],
+        default: Type[PrimitiveCaster],
+        multi: Optional[Type[PrimitiveCaster]] = None,
+    ):
+        self.casters = casters
+        self.default = default
+        self.multi = multi
+
+    def get_types(self) -> List[str]:
+        return list(self.casters.keys())
+
+    def get_caster(
+        self,
+        schema_type: Optional[Union[Iterable[str], str]],
+    ) -> Type["PrimitiveCaster"]:
+        if schema_type is None:
+            return self.default
+        if isinstance(schema_type, Iterable) and not isinstance(
+            schema_type, str
+        ):
+            if self.multi is None:
+                raise TypeError("caster does not accept multiple types")
+            return self.multi
+
+        return self.casters[schema_type]
+
+
+class SchemaCaster:
+    def __init__(
+        self,
+        schema: SchemaPath,
+        schema_validator: SchemaValidator,
+        types_caster: TypesCaster,
+    ):
+        self.schema = schema
+        self.schema_validator = schema_validator
+
+        self.types_caster = types_caster
+
+    def cast(self, value: Any) -> Any:
+        # skip casting for nullable in OpenAPI 3.0
+        if value is None and self.schema.getkey("nullable", False):
+            return value
+
+        schema_type = self.schema.getkey("type")
+        type_caster = self.get_type_caster(schema_type)
+        return type_caster(value)
+
+    def get_type_caster(
+        self,
+        schema_type: Optional[Union[Iterable[str], str]],
+    ) -> PrimitiveCaster:
+        caster_cls = self.types_caster.get_caster(schema_type)
+        return caster_cls(
+            self.schema,
+            self.schema_validator,
+            self,
+        )
+
+    def evolve(self, schema: SchemaPath) -> "SchemaCaster":
+        cls = self.__class__
+
+        return cls(
+            schema,
+            self.schema_validator.evolve(schema),
+            self.types_caster,
+        )

openapi_core/casting/schemas/factories.py

@@ -1,38 +1,35 @@
 from typing import Dict
+from typing import Optional
 
 from jsonschema_path import SchemaPath
 
-from openapi_core.casting.schemas.casters import ArrayCaster
-from openapi_core.casting.schemas.casters import BaseSchemaCaster
-from openapi_core.casting.schemas.casters import CallableSchemaCaster
-from openapi_core.casting.schemas.casters import DummyCaster
+from openapi_core.casting.schemas.casters import SchemaCaster
+from openapi_core.casting.schemas.casters import TypesCaster
 from openapi_core.casting.schemas.datatypes import CasterCallable
 from openapi_core.util import forcebool
+from openapi_core.validation.schemas.datatypes import FormatValidatorsDict
+from openapi_core.validation.schemas.factories import SchemaValidatorsFactory
 
 
 class SchemaCastersFactory:
-    DUMMY_CASTERS = [
-        "string",
-        "object",
-        "any",
-    ]
-    PRIMITIVE_CASTERS: Dict[str, CasterCallable] = {
-        "integer": int,
-        "number": float,
-        "boolean": forcebool,
-    }
-    COMPLEX_CASTERS = {
-        "array": ArrayCaster,
-    }
-
-    def create(self, schema: SchemaPath) -> BaseSchemaCaster:
-        schema_type = schema.getkey("type", "any")
-
-        if schema_type in self.DUMMY_CASTERS:
-            return DummyCaster(schema)
-
-        if schema_type in self.PRIMITIVE_CASTERS:
-            caster_callable = self.PRIMITIVE_CASTERS[schema_type]
-            return CallableSchemaCaster(schema, caster_callable)
-
-        return ArrayCaster(schema, self)
+    def __init__(
+        self,
+        schema_validators_factory: SchemaValidatorsFactory,
+        types_caster: TypesCaster,
+    ):
+        self.schema_validators_factory = schema_validators_factory
+        self.types_caster = types_caster
+
+    def create(
+        self,
+        schema: SchemaPath,
+        format_validators: Optional[FormatValidatorsDict] = None,
+        extra_format_validators: Optional[FormatValidatorsDict] = None,
+    ) -> SchemaCaster:
+        schema_validator = self.schema_validators_factory.create(
+            schema,
+            format_validators=format_validators,
+            extra_format_validators=extra_format_validators,
+        )
+
+        return SchemaCaster(schema, schema_validator, self.types_caster)