Merge pull request #32 from eadwinCode/controller_response_refactor

Throttled Controller

Author: eadwinCode

ninja_extra/constants.py

@@ -7,3 +7,4 @@
 OPTIONS = "OPTIONS"
 TRACE = "TRACE"
 ROUTE_METHODS = [POST, PUT, PATCH, DELETE, GET, HEAD, OPTIONS, TRACE]
+THROTTLED_FUNCTION = "__throttled_endpoint__"

ninja_extra/controllers/base.py

@@ -29,6 +29,7 @@
 from ninja.signature import is_async
 from ninja.utils import normalize_path
 
+from ninja_extra.constants import THROTTLED_FUNCTION
 from ninja_extra.exceptions import APIException, NotFound, PermissionDenied, bad_request
 from ninja_extra.helper import get_function_name
 from ninja_extra.operation import ControllerPathView, Operation
@@ -44,10 +45,11 @@
 from .response import Detail, Id, Ok
 from .route.route_functions import AsyncRouteFunction, RouteFunction
 
-if TYPE_CHECKING:
-    from ninja_extra import NinjaExtraAPI  # pragma: no cover
+if TYPE_CHECKING:  # pragma: no cover
+    from ninja_extra import NinjaExtraAPI
+    from ninja_extra.throttling import BaseThrottle
 
-    from .route.context import RouteContext  # pragma: no cover
+    from .route.context import RouteContext
 
 
 class MissingAPIControllerDecoratorException(Exception):
@@ -115,6 +117,8 @@ def some_method_name(self):
     # `context` variable will change based on the route function called on the APIController
     # that way we can get some specific items things that belong the route function during execution
     context: Optional["RouteContext"] = None
+    throttling_classes: List["BaseThrottle"] = []
+    throttling_init_kwargs: Optional[Dict[Any, Any]] = None
 
     Ok = Ok
     Id = Id
@@ -310,13 +314,21 @@ def tags(self, value: Union[str, List[str], None]) -> None:
         self._tags = tag
 
     def __call__(self, cls: Type) -> Type["ControllerBase"]:
+        from ninja_extra.throttling import throttle
+
         self.auto_import = getattr(cls, "auto_import", self.auto_import)
         if not issubclass(cls, ControllerBase):
             # We force the cls to inherit from `ControllerBase` by creating another type.
             cls = type(cls.__name__, (ControllerBase, cls), {"_api_controller": self})
         else:
             cls._api_controller = self
 
+        assert isinstance(
+            cls.throttling_classes, (list, tuple)
+        ), f"Controller[{cls.__name__}].throttling_class must be a list or tuple"
+        has_throttling_classes = len(cls.throttling_classes) > 0
+        throttling_init_kwargs = cls.throttling_init_kwargs or {}
+
         if not self.tags:
             tag = str(cls.__name__).lower().replace("controller", "")
             self.tags = [tag]
@@ -328,6 +340,11 @@ def __call__(self, cls: Type) -> Type["ControllerBase"]:
                 compute_api_route_function(base_cls, self)
 
         for _, v in self._controller_class_route_functions.items():
+            throttled_endpoint = v.as_view.__dict__.get(THROTTLED_FUNCTION)
+            if not throttled_endpoint and has_throttling_classes:
+                v.route.view_func = throttle(
+                    *cls.throttling_classes, **throttling_init_kwargs
+                )(v.route.view_func)
             self._add_operation_from_route_function(v)
 
         if not is_decorated_with_inject(cls.__init__):

ninja_extra/controllers/response.py

@@ -1,58 +1,146 @@
-from typing import Any, Dict, List, Optional, Type, Union
+import sys
+from typing import (
+    Any,
+    Dict,
+    Generic,
+    Optional,
+    Tuple,
+    Type,
+    TypeVar,
+    Union,
+    no_type_check,
+)
 
 from ninja import Schema
-from pydantic.types import UUID1, UUID3, UUID4, UUID5
-
-from ninja_extra import status
-
-
-class ControllerResponseMeta(type):
-    pass
 
+from .. import status
+from ..schemas import DetailSchema, IdSchema, OkSchema
+
+T = TypeVar("T")
+SCHEMA_KEY = "_schema"
+
+if sys.version_info < (3, 7):  # pragma: no cover
+    from typing import GenericMeta
+
+    class ControllerResponseMeta(GenericMeta):
+        @no_type_check
+        def __new__(mcls, name: str, bases: Tuple, namespace: Dict, **kwargs: Any):
+            args = kwargs.get("args")
+            if args and args[0].__name__ != "T":
+                t = args[0]
+                origin_schema = namespace.get(SCHEMA_KEY)
+                if origin_schema and hasattr(origin_schema, "__generic_model__"):
+                    schema = origin_schema.__generic_model__[t]
+                else:
+                    schema = origin_schema[t]
+                namespace[SCHEMA_KEY] = schema
+            res = super().__new__(mcls, name, bases, namespace, **kwargs)
+            return res
+
+    class GenericControllerResponse(metaclass=ControllerResponseMeta):
+        @no_type_check
+        def __new__(
+            cls: Type["ControllerResponse[T]"], *args: Any, **kwargs: Any
+        ) -> "ControllerResponse[T]":
+
+            if cls._gorg is Generic or "_schema" not in cls.__dict__:
+                raise TypeError(
+                    "Type Generic cannot be instantiated; "
+                    "it can be used only as a base class"
+                )
+            return object.__new__(cls)
+
+else:
+    _generic_types_cache: Dict[
+        Tuple[Type[Any], Union[Any, Tuple[Any, ...]]], Type["ControllerResponse"]
+    ] = {}
+    GenericControllerResponseT = TypeVar(
+        "GenericControllerResponseT", bound="GenericControllerResponse"
+    )
+
+    class ControllerResponseMeta(type):
+        pass
 
-class ControllerResponse(metaclass=ControllerResponseMeta):
+    class GenericControllerResponse(metaclass=ControllerResponseMeta):
+        def __new__(
+            cls: Type["ControllerResponse[T]"], *args: Any, **kwargs: Any
+        ) -> "ControllerResponse[T]":
+
+            if "_schema" not in cls.__dict__:
+                raise TypeError(
+                    "Type Generic cannot be instantiated; "
+                    "it can be used only as a base class"
+                )
+            return object.__new__(cls)
+
+        @no_type_check
+        def __class_getitem__(cls: Type[GenericControllerResponseT], item: Any) -> Any:
+            if isinstance(item, tuple):
+                raise TypeError(
+                    "Cannot parameterize a concrete instantiation of a generic model"
+                )
+
+            _key = (cls, item)
+            _cached_value = _generic_types_cache.get(_key)
+            if _cached_value:
+                return _cached_value
+
+            if str(type(item)) == "<class 'typing.TypeVar'>":
+                result = super().__class_getitem__(item)
+            else:
+                new_schema = cls.__dict__.get(SCHEMA_KEY)
+                if hasattr(new_schema, "__generic_model__"):
+                    new_schema = new_schema.__generic_model__[item]
+
+                result = type(
+                    f"{cls.__name__}[{item.__name__}]", (cls,), {SCHEMA_KEY: new_schema}
+                )
+            _generic_types_cache[_key] = result
+            return result
+
+
+class ControllerResponse(GenericControllerResponse, Generic[T]):
     status_code: int = status.HTTP_204_NO_CONTENT
-
-    def __init__(self, **kwargs: Any) -> None:
-        pass
+    _schema: Optional[T]
 
     @classmethod
-    def get_schema(cls) -> Union[Schema, Type[Schema], Any]:
+    def get_schema(cls) -> Union[Schema, Type[Schema], Any]:  # pragma: no cover
         raise NotImplementedError
 
-    def convert_to_schema(self) -> Any:
+    def convert_to_schema(self) -> Any:  # pragma: no cover
         raise NotImplementedError
 
 
-class Id(ControllerResponse):
+class Id(ControllerResponse[T]):
     """
     Creates a 201 response with id information
     {
         id: int| str| UUID4| UUID1| UUID3| UUID5,
     }
     Example:
            Id(423) ==> 201, {id: 423}
+           OR
+           Id[int](424) ==> 201, {id: 424}
+           Id[UUID4]("883a1a3d-7b10-458d-bccc-f9b7219342c9")
+                ==> 201, {id: "883a1a3d-7b10-458d-bccc-f9b7219342c9"}
     """
 
+    _schema = IdSchema[Any]  # type: ignore
     status_code: int = status.HTTP_201_CREATED
-    id: Union[int, str, UUID4, UUID1, UUID3, UUID5, Any]
 
-    def __init__(self, id: Any) -> None:
+    def __init__(self, id: T) -> None:
         super(Id, self).__init__()
         self.id = id
 
-    class Id(Schema):
-        id: Any
-
     def convert_to_schema(self) -> Any:
-        return self.Id.from_orm(self)
+        return self._schema.from_orm(self)
 
     @classmethod
     def get_schema(cls) -> Union[Schema, Type[Schema], Any]:
-        return cls.Id
+        return cls._schema
 
 
-class Ok(ControllerResponse):
+class Ok(ControllerResponse[T]):
     """
     Creates a 200 response with a detail information.
     {
@@ -61,53 +149,57 @@ class Ok(ControllerResponse):
 
     Example:
        Ok('Saved Successfully') ==> 200, {detail: 'Saved Successfully'}
+       OR
+       class ASchema(BaseModel):
+            name: str
+            age: int
 
+       OK[ASchema](ASchema(name='Eadwin', age=18)) ==> 200, {detail: {'name':'Eadwin', 'age': 18}}
     """
 
     status_code: int = status.HTTP_200_OK
-    detail: Union[str, List[Dict], List[str], Dict] = "Action was successful"
+    _schema = OkSchema[Any]  # type: ignore
 
     def __init__(self, message: Optional[Any] = None) -> None:
         super(Ok, self).__init__()
-        self.detail = message or self.detail
-
-    class Ok(Schema):
-        detail: Union[str, List[Dict], List[str], Dict]
+        self.detail = message or "Action was successful"
 
     def convert_to_schema(self) -> Any:
-        return self.Ok.from_orm(self)
+        return self._schema.from_orm(self)
 
     @classmethod
     def get_schema(cls) -> Union[Schema, Type[Schema], Any]:
-        return cls.Ok
+        return cls._schema
 
 
-class Detail(ControllerResponse):
+class Detail(ControllerResponse[T]):
     """
     Creates a custom response with detail information
     {
         detail: str| List[Dict] | List[str] | Dict,
     }
     Example:
        Detail('Invalid Request', 404) ==> 404, {detail: 'Invalid Request'}
+       OR
+        class ErrorSchema(BaseModel):
+            message: str
+
+       Detail[ErrorSchema](dict(message='Bad Request'),400) ==> 400, {detail: {'message':'Bad Request'}}
     """
 
     status_code: int = status.HTTP_200_OK
-    detail: Union[str, List[Dict], List[str], Dict] = dict()
+    _schema = DetailSchema[Any]  # type: ignore
 
     def __init__(
         self, message: Optional[Any] = None, status_code: int = status.HTTP_200_OK
     ) -> None:
         super(Detail, self).__init__()
-        self.detail = message or self.detail
+        self.detail = message or "Action was successful"
         self.status_code = status_code or self.status_code
 
-    class Detail(Schema):
-        detail: Union[str, List[Dict], List[str], Dict]
-
     def convert_to_schema(self) -> Any:
-        return self.Detail.from_orm(self)
+        return self._schema.from_orm(self)
 
     @classmethod
     def get_schema(cls) -> Union[Schema, Type[Schema], Any]:
-        return cls.Detail
+        return cls._schema

ninja_extra/generic.py

@@ -1,32 +1,47 @@
-from typing import Any, Dict, no_type_check
+from typing import Any, Dict, Optional, Tuple, Type, cast, no_type_check
+
+_generic_types_registry: Dict[Tuple[Type["GenericType"], Type], Type] = {}
 
 
 class GenericModelMeta(type):
-    registry: Dict = {}
+    def _get_name(self: Type["GenericType"], wrap_type: Type) -> str:  # type: ignore
+        _name = f"{self._generic_base_name or self.__name__}"
+        if hasattr(wrap_type, "__name__"):
+            _name += f"[{str(wrap_type.__name__)}]"
+        return _name
+
+    def __getitem__(self: Type["GenericType"], wraps: Any) -> Any:  # type: ignore
+        if (self, wraps) not in _generic_types_registry:
+            new_generic_type = cast(Type, self().get_generic_type(wraps))
+            new_generic_type.__generic_model__ = self
+            new_generic_type.__name__ = self._get_name(wraps)
 
-    def __getitem__(self, wraps: Any) -> Any:
-        if (self, wraps) not in self.__class__.registry:
-            self.__class__.registry[self, wraps] = self().get_generic_type(wraps)
-        return self.__class__.registry[self, wraps]
+            _generic_types_registry[self, wraps] = new_generic_type
+        return _generic_types_registry[self, wraps]
 
 
 @no_type_check
 class GenericType(metaclass=GenericModelMeta):
+    _generic_base_name: Optional[str] = None
     """
     Get a wrapper class that mimic python 3.8 generic support to python3.6, 3.7
     Examples
     --------
     Create a Class to reference the generic model to be create:
-    >>> class ObjectA(GenericType):
+    >>> class ObjectA(GenericType, generic_base_name="A"):
     >>>     def get_generic_type(self, wrap_type):
     >>>         class ObjectAGeneric(self):
     >>>             item: wrap_type
-    >>>         ObjectAGeneric.__name__ = (
-    >>>             f"{self.__class__.__name__}[{str(wrap_type.__name__).capitalize()}]"
-    >>>         )
     >>>         return ObjectAGeneric
 
     Usage:
     >>> class Pass: ...
     >>> object_generic_type  = ObjectA[Pass]
     """
+
+    def __init_subclass__(cls, **kwargs: Any) -> None:
+        cls._generic_base_name = kwargs.get("generic_base_name")
+        return super().__init_subclass__()
+
+    def get_generic_type(self, wrap_type: Any) -> Any:  # pragma: no cover
+        raise NotImplementedError

ninja_extra/schemas/__init__.py

@@ -1,7 +1,17 @@
 from .response import (
+    DetailSchema,
+    IdSchema,
     NinjaPaginationResponseSchema,
+    OkSchema,
     PaginatedResponseSchema,
     RouteParameter,
 )
 
-__all__ = ["PaginatedResponseSchema", "RouteParameter", "NinjaPaginationResponseSchema"]
+__all__ = [
+    "PaginatedResponseSchema",
+    "RouteParameter",
+    "NinjaPaginationResponseSchema",
+    "IdSchema",
+    "OkSchema",
+    "DetailSchema",
+]