fix: generate_crud bug

Author: freemindcore

easy/controller/admin_auto_api.py

@@ -28,7 +28,7 @@ def create_api_controller(
             "model_fields": "__all__",
             "model_recursive": False,
             "model_join": True,
-            "sensitive_fields": ["password"],
+            "sensitive_fields": ["password", "token"],
         },
     )
 

easy/controller/meta.py

@@ -53,47 +53,24 @@ def __init__(self, service=None):  # type: ignore
             self.service = BaseService(model=self.model)
         super().__init__(model=self.model)
 
-    # Define Controller APIs for auto generation
-    async def get_obj(self, request: HttpRequest, id: int) -> Any:
-        """
-        GET /{id}
-        Retrieve a single Object
-        """
-        try:
-            qs = await self.service.get_obj(id)
-        except Exception as e:  # pragma: no cover
-            logger.error(f"Get Error - {e}", exc_info=True)
-            return BaseApiResponse(str(e), message="Get Failed", errno=500)
-        if qs:
-            return qs
-        else:
-            return BaseApiResponse(message="Not Found", errno=404)
-
-    async def del_obj(self, request: HttpRequest, id: int) -> Any:
-        """
-        DELETE /{id}
-        Delete a single Object
-        """
-        if await self.service.del_obj(id):
-            return BaseApiResponse("Deleted.", errno=204)
-        else:
-            return BaseApiResponse("Not Found.", errno=404)
-
-    @paginate
-    async def get_objs(self, request: HttpRequest, filters: str = None) -> Any:
-        """
-        GET /?maximum={int}&filters={filters_dict}
-        Retrieve multiple Object (optional: maximum # and filters)
-        """
-        if filters:
-            return await self.service.get_objs(**json.loads(filters))
-        return await self.service.get_objs()
-
 
 class CrudApiMetaclass(ABCMeta):
     def __new__(mcs, name: str, bases: Tuple[Type[Any], ...], attrs: dict) -> Any:
+        def is_private_attrs(attr_name: str) -> Optional[Match[str]]:
+            return re.match(r"^__[^\d\W]\w*\Z__$", attr_name, re.UNICODE)
+
+        parent_attrs = ChainMap(
+            *[attrs]
+            + [
+                {k: v for (k, v) in vars(base).items() if not (is_private_attrs(k))}
+                for base in bases
+            ]
+        )
+        base_cls_attrs: dict = {}
+        base_cls_attrs.update(parent_attrs)
+
         # Get configs from Meta
-        temp_cls: Type = super().__new__(mcs, name, (object,), attrs)
+        temp_cls: Type = super().__new__(mcs, name, (object,), base_cls_attrs)
         temp_opts: ModelOptions = ModelOptions(getattr(temp_cls, "Meta", None))
         opts_model: Optional[Type[models.Model]] = temp_opts.model
         opts_generate_crud: Optional[bool] = temp_opts.generate_crud
@@ -105,33 +82,56 @@ def __new__(mcs, name: str, bases: Tuple[Type[Any], ...], attrs: dict) -> Any:
             Union[str, List[str]]
         ] = temp_opts.sensitive_fields
 
-        def is_private_attrs(attr_name: str) -> Optional[Match[str]]:
-            return re.match(r"^__[^\d\W]\w*\Z__$", attr_name, re.UNICODE)
+        # Define Controller APIs for auto generation
+        async def get_obj(self, request: HttpRequest, id: int) -> Any:  # type: ignore
+            """
+            GET /{id}
+            Retrieve a single Object
+            """
+            try:
+                qs = await self.service.get_obj(id)
+            except Exception as e:  # pragma: no cover
+                logger.error(f"Get Error - {e}", exc_info=True)
+                return BaseApiResponse(str(e), message="Get Failed", errno=500)
+            if qs:
+                return qs
+            else:
+                return BaseApiResponse(message="Not Found", errno=404)
+
+        async def del_obj(self, request: HttpRequest, id: int) -> Any:  # type: ignore
+            """
+            DELETE /{id}
+            Delete a single Object
+            """
+            if await self.service.del_obj(id):
+                return BaseApiResponse("Deleted.", errno=204)
+            else:
+                return BaseApiResponse("Not Found.", errno=404)
+
+        @paginate
+        async def get_objs(self, request: HttpRequest, filters: str = None) -> Any:  # type: ignore
+            """
+            GET /?maximum={int}&filters={filters_dict}
+            Retrieve multiple Object (optional: maximum # and filters)
+            """
+            if filters:
+                return await self.service.get_objs(**json.loads(filters))
+            return await self.service.get_objs()
 
-        parent_attrs = ChainMap(
-            *[attrs]
-            + [
-                {k: v for (k, v) in vars(base).items() if not (is_private_attrs(k))}
-                for base in bases
-            ]
-        )
-        base_cls_attrs: dict = {}
-        base_cls_attrs.update(parent_attrs)
-        if opts_generate_crud:
+        if opts_generate_crud and opts_model:
             base_cls_attrs.update(
                 {
                     "get_obj": http_get("/{id}", summary="Get a single object")(
-                        copy_func(CrudAPI.get_obj)  # type: ignore
+                        copy_func(get_obj)  # type: ignore
                     ),
                     "del_obj": http_delete("/{id}", summary="Delete a single object")(
-                        copy_func(CrudAPI.del_obj)  # type: ignore
+                        copy_func(del_obj)  # type: ignore
                     ),
                     "get_objs": http_get("/", summary="Get multiple objects")(
-                        copy_func(CrudAPI.get_objs)  # type: ignore
+                        copy_func(get_objs)  # type: ignore
                     ),
                 }
             )
-        if opts_generate_crud and opts_model:
 
             class DataSchema(ModelSchema):
                 class Config:
@@ -173,16 +173,13 @@ async def patch_obj(  # type: ignore
                 f"{opts_model.__name__}__AutoSchema({str(uuid.uuid4())[:4]})"
             )
 
-            setattr(CrudAPI, "patch_obj", patch_obj)
-            setattr(CrudAPI, "add_obj", add_obj)
-
             base_cls_attrs.update(
                 {
                     "patch_obj": http_patch("/{id}", summary="Patch a single object")(
-                        copy_func(CrudAPI.patch_obj)  # type: ignore
+                        copy_func(patch_obj)  # type: ignore
                     ),
                     "add_obj": http_put("/", summary="Create")(
-                        copy_func(CrudAPI.add_obj)  # type: ignore
+                        copy_func(add_obj)  # type: ignore
                     ),
                 }
             )

tests/demo_app/controllers.py

@@ -186,3 +186,14 @@ def __init__(self, service: EventService):
 
     class Meta:
         model = Event
+
+
+@api_controller("unittest", permissions=[AdminSitePermission])
+class NoCrudAPIController(CrudAPIController):
+    def __init__(self, service: EventService):
+        super().__init__(service)
+        self.service = service
+
+    class Meta:
+        model = Event
+        generate_crud = False

tests/demo_app/test_async_auto_crud_apis.py

@@ -11,6 +11,7 @@
     AutoGenCrudSomeFieldsAPIController,
     EventSchema,
     InheritedRecursiveAPIController,
+    NoCrudAPIController,
     RecursiveAPIController,
 )
 from tests.demo_app.models import Category, Client, Event, Type
@@ -25,6 +26,19 @@
 @pytest.mark.skipif(django.VERSION < (3, 1), reason="requires django 3.1 or higher")
 @pytest.mark.django_db
 class TestAutoCrudAdminAPI:
+    async def test_crud_generate_or_not(self, transactional_db, easy_api_client):
+        client = easy_api_client(NoCrudAPIController)
+
+        object_data = dummy_data.copy()
+        object_data.update(title=f"{object_data['title']}_get")
+
+        event = await sync_to_async(Event.objects.create)(**object_data)
+
+        with pytest.raises(Exception):
+            await client.get(
+                f"/{event.id}",
+            )
+
     async def test_crud_default_get_all(self, transactional_db, easy_api_client):
         client = easy_api_client(AutoGenCrudAPIController)
 

tests/demo_app/test_auto_api_creation.py

@@ -37,12 +37,7 @@ async def test_auto_apis(transactional_db, easy_api_client):
         if not str(controller_class).endswith("ClientAdminAPIController"):
             continue
         client = easy_api_client(controller_class)
-        response = await client.get(
-            "/",
-            query=dict(
-                maximum=100,
-            ),
-        )
+        response = await client.get("/")
         assert response.status_code == 200
         assert response.json()["data"] == []