refactor: replace AdminAPI with AdminSitePermission

Author: freemindcore

easy/controller/admin_auto_api.py

@@ -4,8 +4,8 @@
 from django.db import models
 from ninja_extra import ControllerBase, api_controller
 
-from easy.controller.base import BaseAdminAPIController
-from easy.permissions import BaseApiPermission
+from easy.controller.base import CrudAPIController
+from easy.permissions import AdminSitePermission
 
 logger = logging.getLogger(__name__)
 
@@ -17,7 +17,7 @@ class AdminClass(object):
 def create_admin_controller(
     model: models.Model, app_name: str
 ) -> Union[Type[ControllerBase], Type]:
-    """Create AdminAPI class dynamically"""
+    """Create AdminAPI class dynamically, permission class set to AdminSitePermission"""
     model_name = model.__name__  # type:ignore
     Meta = type(
         "Meta",
@@ -31,7 +31,7 @@ def create_admin_controller(
         type,
         class_name,
         (
-            BaseAdminAPIController,
+            CrudAPIController,
             AdminClass,
         ),
         {
@@ -43,5 +43,5 @@ def create_admin_controller(
     return api_controller(
         f"/{app_name}/{model_name.lower()}",
         tags=[f"{model_name} AdminAPI"],
-        permissions=[BaseApiPermission],
+        permissions=[AdminSitePermission],
     )(auto_cls)

easy/controller/base.py

@@ -3,19 +3,12 @@
 
 from ninja_extra import ControllerBase
 
-from easy.controller.meta import AdminApiMetaclass, CrudAPI, CrudApiMetaclass
+from easy.controller.meta import CrudAPI, CrudApiMetaclass
 from easy.services import BaseService
 
 logger = logging.getLogger(__name__)
 
 
-class BaseAdminAPIController(ControllerBase, CrudAPI, metaclass=AdminApiMetaclass):
-    """For AdminAPI"""
-
-    def __init__(self, service: Union["BaseService", None] = None):
-        super().__init__(service=service)  # pragma: no cover
-
-
 class CrudAPIController(ControllerBase, CrudAPI, metaclass=CrudApiMetaclass):
     """For Client facing APIs"""
 

easy/controller/meta.py

@@ -85,18 +85,8 @@ async def filter_exclude_objs(self, filters: Union[str, bytes]) -> Any:
     #     return await self.service.recover_obj()
 
 
-class AdminCrudAPI(CrudAPI):
-    ...
-
-
 class CrudApiMetaclass(ABCMeta):
     def __new__(mcs, name: str, bases: Tuple[Type[Any], ...], attrs: dict) -> Any:
-        return mcs.generate_cls(mcs, name, (ControllerBase, CrudAPI), attrs)
-
-    @staticmethod
-    def generate_cls(
-        mcs: Type[Any], name: str, bases: Tuple[Type[Any], ...], attrs: dict
-    ) -> Type[Any]:
         # Get configs from Meta
         temp_base: Type = type.__new__(type, "object", (), {})
         temp_cls: Type = super(CrudApiMetaclass, mcs).__new__(
@@ -177,7 +167,9 @@ async def patch_obj(  # type: ignore
                 }
             )
 
-        new_base: Type = type.__new__(type, name, bases, base_cls_attrs)
+        new_base: Type = type.__new__(
+            type, name, (ControllerBase, CrudAPI), base_cls_attrs
+        )
         new_cls: Type = super(CrudApiMetaclass, mcs).__new__(
             mcs, name, (new_base,), attrs
         )
@@ -190,11 +182,6 @@ async def patch_obj(  # type: ignore
         return new_cls
 
 
-class AdminApiMetaclass(CrudApiMetaclass):
-    def __new__(mcs, name: str, bases: Union[Type], attrs: dict) -> Any:
-        return mcs.generate_cls(mcs, name, (ControllerBase, AdminCrudAPI), attrs)
-
-
 class ModelOptions:
     def __init__(self, options: object = None):
         self.model: Optional[Type[models.Model]] = getattr(options, "model", None)

easy/main.py

@@ -152,41 +152,3 @@ def create_response(
             status=status,
             temporal_response=temporal_response,
         )  # pragma: no cover
-
-
-class EasyAdminAPI(EasyAPI):
-    def __init__(
-        self,
-        *,
-        title: str = "Easy AdminAPI",
-        version: str = "1.0.0",
-        description: str = "",
-        openapi_url: Optional[str] = "/openapi.json",
-        docs_url: Optional[str] = "/docs",
-        urls_namespace: Optional[str] = None,
-        csrf: bool = False,
-        auth: Union[
-            Sequence[Callable[..., Any]], Callable[..., Any], NOT_SET_TYPE, None
-        ] = NOT_SET,
-        renderer: Optional[BaseRenderer] = EasyJSONRenderer(),
-        parser: Optional[Parser] = None,
-        app_name: str = "ninja",
-        easy_extra: bool = True,
-        easy_output: bool = True,
-    ) -> None:
-        super(EasyAdminAPI, self).__init__(
-            title=title,
-            version=version,
-            description=description,
-            openapi_url=openapi_url,
-            docs_url=docs_url,
-            urls_namespace=urls_namespace,
-            csrf=csrf,
-            auth=auth,
-            renderer=renderer,
-            parser=parser,
-            app_name=app_name,
-            easy_output=easy_output,
-        )
-        self.easy_extra = easy_extra
-        self.easy_output = easy_output

easy/permissions/__init__.py

@@ -5,14 +5,16 @@
     IsAuthenticatedOrReadOnly,
 )
 
+from .adminsite import AdminSitePermission
 from .base import BaseApiPermission
 from .superuser import IsSuperUser
 
 __all__ = [
-    "BaseApiPermission",
-    "IsSuperUser",
     "AllowAny",
+    "AdminSitePermission",
+    "BaseApiPermission",
     "IsAuthenticated",
-    "IsAdminUser",
     "IsAuthenticatedOrReadOnly",
+    "IsAdminUser",
+    "IsSuperUser",
 ]

easy/permissions/adminsite.py

@@ -0,0 +1,82 @@
+from typing import TYPE_CHECKING
+
+from django.http import HttpRequest
+from ninja_extra.permissions import IsAdminUser
+
+if TYPE_CHECKING:
+    from ninja_extra.controllers.base import ControllerBase  # pragma: no cover
+
+
+class AdminSitePermission(IsAdminUser):
+    """
+    Allows access only to super user.
+    """
+
+    def has_permission(
+        self, request: HttpRequest, controller: "ControllerBase"
+    ) -> bool:
+        """
+        Return `True` if permission is granted, `False` otherwise.
+        """
+        user = request.user or request.auth  # type: ignore
+        has_perm: bool = True
+        if request.method == "DELETE":
+            has_perm = bool(user.is_superuser)  # type: ignore
+        if request.method in ("PUT", "PATCH", "POST"):
+            has_perm = bool(user.is_staff or user.is_superuser)  # type: ignore
+        return has_perm and super().has_permission(request, controller)
+
+
+#
+# class AdminSitePermissionV2(AdminSitePermission):
+#     def has_permission(
+#         self, request: HttpRequest, controller: "ControllerBase"
+#     ) -> bool:
+#         """
+#         Return `True` if permission is granted, `False` otherwise.
+#         """
+#         user = request.user
+#         has_perm = True
+#         if hasattr(controller, "model"):
+#             model = controller.model
+#             app = model._meta.app_label
+#             has_perm = user.has_perm(f"{app}.view_{model.__name__}")
+#             if request.method in ("PUT",):
+#                 has_perm = user.has_perm(f"{app}.add_{model.__name__}")
+#             if request.method in ("PATCH", "POST"):
+#                 has_perm = user.has_perm(f"{app}.change_{model.__name__}")
+#             if request.method in ("DELETE",):
+#                 has_perm = user.has_perm(f"{app}.delete_{model.__name__}")
+#         if user.is_superuser:
+#             has_perm = True
+#         return bool(user and user.is_authenticated and user.is_active and has_perm)
+#
+#     async def has_permission(
+#         self, request: HttpRequest, controller: "ControllerBase"
+#     ) -> bool:
+#         """
+#         Return `True` if permission is granted, `False` otherwise.
+#         """
+#         user = request.user
+#         has_perm = False
+#         if hasattr(controller, "model"):
+#             model = controller.model
+#             app = model._meta.app_label
+#             has_perm = await sync_to_async(user.has_perm)(
+#                 f"{app}.view_{model.__name__}"
+#             )
+#             if request.method in ("PUT",):
+#                 has_perm = await sync_to_async(user.has_perm)(
+#                     f"{app}.add_{model.__name__}"
+#                 )
+#             if request.method in ("PATCH", "POST"):
+#                 has_perm = await sync_to_async(user.has_perm)(
+#                     f"{app}.change_{model.__name__}"
+#                 )
+#             if request.method in ("DELETE",):
+#                 has_perm = await sync_to_async(user.has_perm)(
+#                     f"{app}.delete_{model.__name__}"
+#                 )
+#         if user.is_superuser:
+#             has_perm = True
+#         return bool(user and user.is_authenticated and user.is_active and has_perm)

easy/services/permission.py

@@ -1,30 +1,25 @@
 import logging
-from typing import TYPE_CHECKING, Any, Union
+from typing import TYPE_CHECKING, Any
 
 from django.http import HttpRequest
 
 if TYPE_CHECKING:
-    from django.contrib.auth.models import AbstractUser  # pragma: no cover
     from ninja_extra.controllers.base import ControllerBase  # pragma: no cover
 
 
 logger = logging.getLogger(__name__)
 
 
 class PermissionService(object):
+    """Base permission service for extra customization needs"""
+
     def check_permission(
         self, request: HttpRequest, controller: "ControllerBase"
     ) -> bool:
         """
         Return `True` if permission is granted, `False` otherwise.
         """
-        user: Union[AbstractUser] = request.user  # type: ignore
-        has_perm: bool = True
-        if request.method == "DELETE":
-            has_perm = bool(user.is_superuser)
-        if request.method in ("PUT", "PATCH", "POST"):
-            has_perm = bool(user.is_staff or user.is_superuser)
-        return bool(user and user.is_authenticated and user.is_active and has_perm)
+        return True
 
     def check_object_permission(
         self, request: HttpRequest, controller: "ControllerBase", obj: Any
@@ -33,56 +28,3 @@ def check_object_permission(
         Return `True` if permission is granted, `False` otherwise.
         """
         return True
-
-    #
-    # def check_permission_v2(
-    #     self, request: HttpRequest, controller: "ControllerBase"
-    # ) -> bool:
-    #     """
-    #     Return `True` if permission is granted, `False` otherwise.
-    #     """
-    #     user = request.user
-    #     has_perm = True
-    #     if hasattr(controller, "model"):
-    #         model = controller.model
-    #         app = model._meta.app_label
-    #         has_perm = user.has_perm(f"{app}.view_{model.__name__}")
-    #         if request.method in ("PUT",):
-    #             has_perm = user.has_perm(f"{app}.add_{model.__name__}")
-    #         if request.method in ("PATCH", "POST"):
-    #             has_perm = user.has_perm(f"{app}.change_{model.__name__}")
-    #         if request.method in ("DELETE",):
-    #             has_perm = user.has_perm(f"{app}.delete_{model.__name__}")
-    #     if user.is_superuser:
-    #         has_perm = True
-    #     return bool(user and user.is_authenticated and user.is_active and has_perm)
-    #
-    # async def async_check_permission(
-    #     self, request: HttpRequest, controller: "ControllerBase"
-    # ) -> bool:
-    #     """
-    #     Return `True` if permission is granted, `False` otherwise.
-    #     """
-    #     user = request.user
-    #     has_perm = False
-    #     if hasattr(controller, "model"):
-    #         model = controller.model
-    #         app = model._meta.app_label
-    #         has_perm = await sync_to_async(user.has_perm)(
-    #             f"{app}.view_{model.__name__}"
-    #         )
-    #         if request.method in ("PUT",):
-    #             has_perm = await sync_to_async(user.has_perm)(
-    #                 f"{app}.add_{model.__name__}"
-    #             )
-    #         if request.method in ("PATCH", "POST"):
-    #             has_perm = await sync_to_async(user.has_perm)(
-    #                 f"{app}.change_{model.__name__}"
-    #             )
-    #         if request.method in ("DELETE",):
-    #             has_perm = await sync_to_async(user.has_perm)(
-    #                 f"{app}.delete_{model.__name__}"
-    #             )
-    #     if user.is_superuser:
-    #         has_perm = True
-    #     return bool(user and user.is_authenticated and user.is_active and has_perm)

tests/demo_app/apis.py

@@ -1,14 +1,14 @@
 from easy.main import EasyAPI
 from tests.demo_app.auth import jwt_auth_async
 from tests.demo_app.controllers import (
-    EasyAdminAPIController,
+    AutoGenCrudAPIController,
     EasyCrudAPIController,
-    EasyCrudBasePermissionAPIController,
+    PermissionAPIController,
 )
 
 api_unittest = EasyAPI(auth=jwt_auth_async)
 api_unittest.register_controllers(
     EasyCrudAPIController,
-    EasyCrudBasePermissionAPIController,
-    EasyAdminAPIController,
+    PermissionAPIController,
+    AutoGenCrudAPIController,
 )

tests/demo_app/conftest.py

@@ -3,8 +3,7 @@
 import pytest
 from django.contrib.auth import get_user_model
 
-from easy.controller.base import BaseAdminAPIController, CrudAPIController
-from easy.main import EasyAdminAPI
+from easy.controller.base import CrudAPIController
 from easy.testing import EasyTestClient
 from tests.demo_app.auth import JWTAuthAsync, jwt_auth_async
 from tests.demo_app.factories import UserFactory
@@ -44,30 +43,3 @@ def create_client(
 
     yield create_client
     setattr(JWTAuthAsync, "__call__", orig_func)
-
-
-@pytest.fixture
-def easy_admin_api_client(user) -> EasyTestClient:
-    """For EasyAdminAPIAuthASync testings"""
-
-    orig_func = copy.deepcopy(JWTAuthAsync.__call__)
-
-    async def mock_func(self, request):
-        setattr(request, "user", user)
-        return True
-
-    setattr(JWTAuthAsync, "__call__", mock_func)
-
-    def create_client(
-        api: BaseAdminAPIController,
-        is_staff: bool = True,
-        is_superuser: bool = False,
-    ):
-        setattr(user, "is_staff", is_staff)
-        setattr(user, "is_superuser", is_superuser)
-
-        client = EasyTestClient(api, auth=jwt_auth_async, api_cls=EasyAdminAPI)
-        return client
-
-    yield create_client
-    setattr(JWTAuthAsync, "__call__", orig_func)