Merge pull request #12 from eadwinCode/openapi_test

Openapi Module Test

Author: eadwinCode

README.md

@@ -108,7 +108,6 @@ Project is still in development
 - Remaining testing modules:
     - configuration
     - guard
-    - openapi
 - Project CLI scaffolding
 - Documentation
 - Database Plugin with [Encode/ORM](https://github.com/encode/orm)

ellar/common/__init__.py

@@ -8,7 +8,7 @@
 from .decorators.html import render, template_filter, template_global
 from .decorators.middleware import middleware
 from .decorators.modules import Module
-from .decorators.openapi import openapi
+from .decorators.openapi import openapi_info
 from .decorators.request import on_shutdown, on_startup
 from .decorators.serializer import serializer_filter
 from .decorators.versioning import version
@@ -48,7 +48,7 @@
     "ParamTypes",
     "set_meta",
     "Controller",
-    "openapi",
+    "openapi_info",
     "version",
     "delete",
     "get",

ellar/common/decorators/__init__.py

@@ -5,7 +5,7 @@
 from .html import render, template_filter, template_global  # noqa
 from .middleware import middleware  # noqa
 from .modules import Module  # noqa
-from .openapi import openapi  # noqa
+from .openapi import openapi_info  # noqa
 from .request import on_shutdown, on_startup  # noqa
 from .serializer import serializer_filter  # noqa
 from .versioning import version  # noqa

ellar/common/decorators/openapi.py

@@ -6,7 +6,7 @@
 from .base import set_meta
 
 
-def openapi(
+def openapi_info(
     operation_id: t.Optional[str] = None,
     summary: t.Optional[str] = None,
     description: t.Optional[str] = None,

ellar/core/modules/helper.py

@@ -1,8 +1,7 @@
+import inspect
 import typing as t
 from functools import wraps
 
-from starlette.routing import iscoroutinefunction_or_partial
-
 
 def _executor_wrapper_async(
     cls: t.Type, func: t.Callable
@@ -25,7 +24,7 @@ def _decorator(*args: t.Any, **kwargs: t.Any) -> t.Any:
 def class_parameter_executor_wrapper(
     cls: t.Type, func: t.Callable
 ) -> t.Union[t.Callable, t.Callable[..., t.Coroutine]]:
-    if iscoroutinefunction_or_partial(func):
+    if inspect.iscoroutinefunction(func):
         return _executor_wrapper_async(cls, func)
 
     return _executor_wrapper(cls, func)

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,34 @@ def __init__(
         )
         self.description = description
         self.meta = kwargs
-        self.schema = schema
 
+        _schema: t.Optional[ResponseModelField] = schema or self.model_schema
+        if _schema and not isinstance(_schema, ResponseModelField):
+            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,
+                ),
+            )
+        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 +151,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,