Refactored and added a factor for creating ResponseModels

Author: eadwinCode

ellar/core/response/model/__init__.py

@@ -5,6 +5,7 @@
     ResponseTypeDefinitionConverter,
     RouteResponseExecution,
 )
+from .factory import create_response_model
 from .html import HTMLResponseModel
 from .interface import IResponseModel
 from .json import EmptyAPIResponseModel, JSONResponseModel
@@ -21,4 +22,5 @@
     "ResponseTypeDefinitionConverter",
     "IResponseModel",
     "HTMLResponseModel",
+    "create_response_model",
 ]

ellar/core/response/model/base.py

@@ -1,5 +1,5 @@
 import typing as t
-from abc import ABC
+from abc import ABC, abstractmethod
 from dataclasses import is_dataclass
 
 from pydantic import BaseModel
@@ -9,6 +9,7 @@
 from ellar.core.context import IExecutionContext
 from ellar.core.converters import TypeDefinitionConverter
 from ellar.exceptions import RequestValidationError
+from ellar.helper.modelfield import create_model_field
 from ellar.reflect import reflect
 from ellar.serializer import (
     BaseSerializer,
@@ -64,7 +65,9 @@ def serialize(
 
 class BaseResponseModel(IResponseModel, ABC):
     __slots__ = ("_response_type", "media_type", "description", "meta", "schema")
+
     response_type: t.Type[Response] = Response
+    model_schema: t.Union[ResponseModelField, t.Any] = None
 
     def __init__(
         self,
@@ -80,17 +83,38 @@ def __init__(
         )
         self.description = description
         self.meta = kwargs
-        self.schema = schema
 
+        _schema: t.Optional[ResponseModelField] = None
+        if (
+            schema
+            or self.model_schema
+            and not isinstance(schema or self.model_schema, ResponseModelField)
+        ):
+            new_response_schema = ResponseTypeDefinitionConverter(
+                schema or self.model_schema
+            ).re_group_outer_type()
+
+            _schema = t.cast(
+                ResponseModelField,
+                create_model_field(
+                    name="response_model",
+                    type_=new_response_schema,
+                    model_field_class=ResponseModelField,
+                ),
+            )
+        self.schema: t.Optional[ResponseModelField] = _schema
+
+    @abstractmethod
     def get_model_field(self) -> t.Optional[t.Union[ResponseModelField, t.Any]]:
-        return self.schema
+        pass
 
+    @abstractmethod
     def serialize(
         self,
         response_obj: t.Any,
         serializer_filter: t.Optional[SerializerFilter] = None,
     ) -> t.Union[t.List[t.Dict], t.Dict, t.Any]:
-        return response_obj
+        pass
 
     def create_response(
         self, context: IExecutionContext, response_obj: t.Any, status_code: int
@@ -131,22 +155,15 @@ def __copy__(self, memodict: t.Dict = {}) -> "BaseResponseModel":
 
 
 class ResponseModel(BaseResponseModel):
-    @classmethod
-    def create_model(
-        cls,
-        *args: t.Any,
-        response_type: t.Type[Response] = None,
-        description: str = "Successful Response",
-        schema: t.Union[t.Type[ResponseModelField], t.Any] = None,
-        **kwargs: t.Any,
-    ) -> "ResponseModel":
+    def get_model_field(self) -> t.Optional[t.Union[ResponseModelField, t.Any]]:
+        return self.schema
 
-        return cls(
-            response_type=response_type,
-            description=description,
-            schema=schema,
-            **kwargs,
-        )
+    def serialize(
+        self,
+        response_obj: t.Any,
+        serializer_filter: t.Optional[SerializerFilter] = None,
+    ) -> t.Union[t.List[t.Dict], t.Dict, t.Any]:
+        return response_obj
 
 
 class RouteResponseExecution(Exception):

ellar/core/response/model/factory.py

@@ -0,0 +1,23 @@
+import typing as t
+
+from starlette.responses import Response
+
+if t.TYPE_CHECKING:
+    from .base import ResponseModel, ResponseModelField
+
+T = t.TypeVar("T")
+
+
+def create_response_model(
+    response_model: t.Union[t.Type["ResponseModel"], t.Type[T]],
+    /,
+    response_type: t.Type[Response] = None,
+    description: str = "Successful Response",
+    schema: t.Union[t.Type["ResponseModelField"], t.Any] = None,
+    **kwargs: t.Any,
+) -> t.Union["ResponseModel", T]:
+    _init_kwargs: t.Dict[str, t.Any] = dict(
+        response_type=response_type, description=description, schema=schema
+    )
+    _init_kwargs.update(kwargs)
+    return response_model(**_init_kwargs)

ellar/core/response/model/interface.py

@@ -14,15 +14,6 @@ class IResponseModel(ABC):
     media_type: str
     description: str
 
-    @classmethod
-    @abstractmethod
-    def create_model(
-        cls,
-        *args: t.Any,
-        **kwargs: t.Any,
-    ) -> "IResponseModel":
-        """Creates ResponseModel Object"""
-
     @abstractmethod
     def get_model_field(self) -> t.Optional[t.Union[ModelField, t.Any]]:
         """Gets Model Fields"""

ellar/core/response/model/json.py

@@ -1,45 +1,26 @@
 import typing as t
 
-from pydantic import BaseModel
-
 from ellar.constants import SERIALIZER_FILTER_KEY
 from ellar.core.context import IExecutionContext
 from ellar.helper.modelfield import create_model_field
 from ellar.reflect import reflect
 from ellar.serializer import SerializerFilter, serialize_object
 
 from ..responses import JSONResponse, Response
-from .base import BaseResponseModel, ResponseModelField, ResponseTypeDefinitionConverter
-
+from .base import ResponseModel, ResponseModelField
 
-class JSONResponseModel(BaseResponseModel):
-    response_type: t.Type[Response] = JSONResponse
-    schema: ResponseModelField
+DictSchema: ResponseModelField = t.cast(
+    ResponseModelField,
+    create_model_field(
+        name="response_model",
+        type_=dict,
+        model_field_class=ResponseModelField,
+    ),
+)
 
-    @classmethod
-    def create_model(  # type: ignore
-        cls,
-        schema: t.Union[t.Type[BaseModel], t.Any],
-        description: str = "Successful Response",
-        **kwargs: t.Any,
-    ) -> "JSONResponseModel":
-        new_response_schema = ResponseTypeDefinitionConverter(
-            schema
-        ).re_group_outer_type()
 
-        _schema = t.cast(
-            ResponseModelField,
-            create_model_field(
-                name="response_model",
-                type_=new_response_schema,
-                model_field_class=ResponseModelField,
-            ),
-        )
-        return cls(
-            description=description,
-            schema=_schema,
-            **kwargs,
-        )
+class JSONResponseModel(ResponseModel):
+    response_type: t.Type[Response] = JSONResponse
 
     def create_response(
         self, context: IExecutionContext, response_obj: t.Any, status_code: int
@@ -64,32 +45,12 @@ def serialize(
         response_obj: t.Any,
         serializer_filter: t.Optional[SerializerFilter] = None,
     ) -> t.Union[t.List[t.Dict], t.Dict, t.Any]:
+        assert self.schema, "schema must exist for JSONResponseModel"
         return self.schema.serialize(response_obj, serializer_filter=serializer_filter)
 
 
 class EmptyAPIResponseModel(JSONResponseModel):
-    @classmethod
-    def create_model(  # type: ignore
-        cls,
-        description: str = "Successful Response",
-        **kwargs: t.Any,
-    ) -> "EmptyAPIResponseModel":
-        new_response_schema = dict
-
-        _schema = t.cast(
-            ResponseModelField,
-            create_model_field(
-                name="response_model",
-                type_=new_response_schema,
-                model_field_class=ResponseModelField,
-            ),
-        )
-        return cls(
-            response_type=JSONResponse,
-            description=description,
-            schema=_schema,
-            **kwargs,
-        )
+    model_schema = DictSchema
 
     def serialize(
         self,

ellar/core/response/model/route.py

@@ -6,6 +6,7 @@
 
 from ..responses import Response
 from .base import ResponseModel, ResponseResolver, RouteResponseExecution
+from .factory import create_response_model
 from .interface import IResponseModel
 from .json import JSONResponseModel
 
@@ -36,16 +37,19 @@ def convert_route_responses_to_response_models(
             if isinstance(response_schema, type) and issubclass(
                 response_schema, Response
             ):
-                self.models[status_code] = ResponseModel.create_model(
-                    response_type=response_schema, description=description
+                self.models[status_code] = create_response_model(
+                    ResponseModel,
+                    response_type=response_schema,
+                    description=description,
                 )
                 continue
 
             if isinstance(response_schema, IResponseModel):
                 self.models[status_code] = response_schema
                 continue
 
-            self.models[status_code] = JSONResponseModel.create_model(
+            self.models[status_code] = create_response_model(
+                JSONResponseModel,
                 schema=t.cast(t.Type[BaseModel], response_schema),
                 description=description,
             )