x-model extension import model class

Author: p1c2u

docs/extensions.rst

@@ -0,0 +1,34 @@
+Extensions
+==========
+
+x-model
+-------
+
+By default, objects are unmarshalled to dynamically created dataclasses. You can use your own dataclasses, pydantic models or models generated by third party generators (i.e. `datamodel-code-generator <https://github.com/koxudaxi/datamodel-code-generator>`__) by providing ``x-model`` property inside schema definition with location of your class.
+
+.. code-block:: yaml
+
+   ...
+   components:
+     schemas:
+       Coordinates:
+         x-model: foo.bar.Coordinates
+         type: object
+         required:
+           - lat
+           - lon
+         properties:
+           lat:
+             type: number
+           lon:
+             type: number
+
+.. code-block:: python
+
+   # foo/bar.py
+   from dataclasses import dataclass
+
+   @dataclass
+   class Coordinates:
+       lat: float
+       lon: float

docs/index.rst

@@ -31,6 +31,7 @@ Table of contents
 
    installation
    usage
+   extensions
    customizations
    integrations
 

openapi_core/contrib/django/handlers.py

@@ -47,7 +47,7 @@ def format_openapi_error(cls, error: Exception) -> Dict[str, Any]:
         return {
             "title": str(error),
             "status": cls.OPENAPI_ERROR_STATUS.get(error.__class__, 400),
-            "class": str(type(error)),
+            "type": str(type(error)),
         }
 
     @classmethod

openapi_core/contrib/falcon/handlers.py

@@ -53,7 +53,7 @@ def format_openapi_error(cls, error: Exception) -> Dict[str, Any]:
         return {
             "title": str(error),
             "status": cls.OPENAPI_ERROR_STATUS.get(error.__class__, 400),
-            "class": str(type(error)),
+            "type": str(type(error)),
         }
 
     @classmethod

openapi_core/extensions/models/factories.py

@@ -1,33 +1,51 @@
 """OpenAPI X-Model extension factories module"""
+from dataclasses import make_dataclass
+from pydoc import ErrorDuringImport
+from pydoc import locate
 from typing import Any
 from typing import Dict
+from typing import Iterable
 from typing import Optional
 from typing import Type
 
-from openapi_core.extensions.models.models import Model
+from openapi_core.extensions.models.types import Field
 
 
-class ModelClassFactory:
+class DictFactory:
 
-    base_class = Model
+    base_class = dict
 
-    def create(self, name: str) -> Type[Model]:
-        return type(name, (self.base_class,), {})
+    def create(self, fields: Iterable[Field]) -> Type[Dict[Any, Any]]:
+        return self.base_class
 
 
-class ModelFactory:
-    def __init__(
-        self, model_class_factory: Optional[ModelClassFactory] = None
-    ):
-        self.model_class_factory = model_class_factory or ModelClassFactory()
-
+class DataClassFactory(DictFactory):
     def create(
-        self, properties: Optional[Dict[str, Any]], name: Optional[str] = None
-    ) -> Model:
-        name = name or "Model"
+        self,
+        fields: Iterable[Field],
+        name: str = "Model",
+    ) -> Type[Any]:
+        return make_dataclass(name, fields, frozen=True)
 
-        model_class = self._create_class(name)
-        return model_class(properties)
 
-    def _create_class(self, name: str) -> Type[Model]:
-        return self.model_class_factory.create(name)
+class ModelClassImporter(DataClassFactory):
+    def create(
+        self,
+        fields: Iterable[Field],
+        name: str = "Model",
+        model: Optional[str] = None,
+    ) -> Any:
+        if model is None:
+            return super().create(fields, name=name)
+
+        model_class = self._get_class(model)
+        if model_class is not None:
+            return model_class
+
+        return super().create(fields, name=model)
+
+    def _get_class(self, model_class_path: str) -> Optional[object]:
+        try:
+            return locate(model_class_path)
+        except ErrorDuringImport:
+            return None

openapi_core/extensions/models/models.py

@@ -0,0 +1,5 @@
+from typing import Any
+from typing import Tuple
+from typing import Union
+
+Field = Union[str, Tuple[str, Any]]

openapi_core/extensions/models/types.py

@@ -3,6 +3,7 @@
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Dict
+from typing import Iterable
 from typing import List
 from typing import Optional
 
@@ -16,7 +17,7 @@
 from openapi_schema_validator._format import oas30_format_checker
 from openapi_schema_validator._types import is_string
 
-from openapi_core.extensions.models.factories import ModelFactory
+from openapi_core.extensions.models.factories import ModelClassImporter
 from openapi_core.schema.schemas import get_all_properties
 from openapi_core.schema.schemas import get_all_properties_names
 from openapi_core.spec import Spec
@@ -196,8 +197,8 @@ class ObjectUnmarshaller(ComplexUnmarshaller):
     }
 
     @property
-    def model_factory(self) -> ModelFactory:
-        return ModelFactory()
+    def object_class_factory(self) -> ModelClassImporter:
+        return ModelClassImporter()
 
     def unmarshal(self, value: Any) -> Any:
         try:
@@ -230,11 +231,11 @@ def _unmarshal_object(self, value: Any) -> Any:
         else:
             properties = self._unmarshal_properties(value)
 
-        if "x-model" in self.schema:
-            name = self.schema["x-model"]
-            return self.model_factory.create(properties, name=name)
+        model = self.schema.getkey("x-model")
+        fields: Iterable[str] = properties and properties.keys() or []
+        object_class = self.object_class_factory.create(fields, model=model)
 
-        return properties
+        return object_class(**properties)
 
     def _unmarshal_properties(
         self, value: Any, one_of_schema: Optional[Spec] = None
@@ -253,17 +254,18 @@ def _unmarshal_properties(
         additional_properties = self.schema.getkey(
             "additionalProperties", True
         )
-        if isinstance(additional_properties, dict):
-            additional_prop_schema = self.schema / "additionalProperties"
+        if additional_properties is not False:
+            # free-form object
+            if additional_properties is True:
+                additional_prop_schema = Spec.from_dict({})
+            # defined schema
+            else:
+                additional_prop_schema = self.schema / "additionalProperties"
             for prop_name in extra_props:
                 prop_value = value[prop_name]
                 properties[prop_name] = self.unmarshallers_factory.create(
                     additional_prop_schema
                 )(prop_value)
-        elif additional_properties is True:
-            for prop_name in extra_props:
-                prop_value = value[prop_name]
-                properties[prop_name] = prop_value
 
         for prop_name, prop in list(all_props.items()):
             read_only = prop.getkey("readOnly", False)

openapi_core/unmarshalling/schemas/unmarshallers.py

@@ -54,7 +54,7 @@ def test_get_no_required_param(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.validation.exceptions."
                         "MissingRequiredParameter'>"
                     ),
@@ -101,7 +101,7 @@ def test_post_server_invalid(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.templating.paths.exceptions."
                         "ServerNotFound'>"
                     ),
@@ -148,7 +148,7 @@ def test_post_required_header_param_missing(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.validation.exceptions."
                         "MissingRequiredParameter'>"
                     ),
@@ -176,7 +176,7 @@ def test_post_media_type_invalid(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.templating.media_types."
                         "exceptions.MediaTypeNotFound'>"
                     ),
@@ -213,7 +213,7 @@ def test_post_required_cookie_param_missing(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.validation.exceptions."
                         "MissingRequiredParameter'>"
                     ),
@@ -267,7 +267,7 @@ def test_get_unauthorized(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.validation.exceptions."
                         "InvalidSecurity'>"
                     ),
@@ -289,7 +289,7 @@ def test_delete_method_invalid(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.templating.paths.exceptions."
                         "OperationNotFound'>"
                     ),

tests/integration/contrib/django/test_django_project.py

@@ -65,7 +65,7 @@ def test_post_server_invalid(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.templating.paths.exceptions."
                         "ServerNotFound'>"
                     ),
@@ -119,7 +119,7 @@ def test_post_required_header_param_missing(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.validation.exceptions."
                         "MissingRequiredParameter'>"
                     ),
@@ -155,7 +155,7 @@ def test_post_media_type_invalid(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.templating.media_types."
                         "exceptions.MediaTypeNotFound'>"
                     ),
@@ -198,7 +198,7 @@ def test_post_required_cookie_param_missing(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.validation.exceptions."
                         "MissingRequiredParameter'>"
                     ),
@@ -249,7 +249,7 @@ def test_get_server_invalid(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.templating.paths.exceptions."
                         "ServerNotFound'>"
                     ),
@@ -283,7 +283,7 @@ def test_get_unauthorized(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.validation.exceptions."
                         "InvalidSecurity'>"
                     ),
@@ -324,7 +324,7 @@ def test_delete_method_invalid(self, client):
         expected_data = {
             "errors": [
                 {
-                    "class": (
+                    "type": (
                         "<class 'openapi_core.templating.paths.exceptions."
                         "OperationNotFound'>"
                     ),