added get_object with check_object_permission support

Author: eadwinCode

ninja_extra/controllers/base.py

@@ -4,28 +4,35 @@
     Any,
     Callable,
     Dict,
-    Iterator,
+    Iterable,
     List,
     Optional,
     Type,
+    Union,
     cast,
     no_type_check,
 )
 
+from django.db.models import Model, QuerySet
+from django.http import HttpResponse
 from injector import inject, is_decorated_with_inject
 from ninja import NinjaAPI
 from ninja.constants import NOT_SET
 from ninja.operation import Operation
 from ninja.security.base import AuthBase
 from ninja.types import DictStrAny
 
-from ninja_extra.exceptions import PermissionDenied
+from ninja_extra.exceptions import APIException, NotFound, PermissionDenied
 from ninja_extra.operation import PathView
 from ninja_extra.permissions import AllowAny, BasePermission
-from ninja_extra.shortcuts import fail_silently
+from ninja_extra.shortcuts import (
+    fail_silently,
+    get_object_or_exception,
+    get_object_or_none,
+)
 from ninja_extra.types import PermissionType
 
-from .response import ControllerResponse, Detail, Id, Ok
+from .response import Detail, Id, Ok
 from .route.route_functions import RouteFunction
 from .router import ControllerRouter
 
@@ -171,7 +178,7 @@ def add_api_operation(
         return operation
 
     @classmethod
-    def get_route_functions(cls) -> Iterator[RouteFunction]:
+    def get_route_functions(cls) -> Iterable[RouteFunction]:
         for method in cls.__dict__.values():
             if isinstance(method, RouteFunction):
                 yield method
@@ -181,7 +188,28 @@ def permission_denied(cls, permission: BasePermission) -> None:
         message = getattr(permission, "message", None)
         raise PermissionDenied(message)
 
-    def get_permissions(self) -> Iterator[BasePermission]:
+    def get_object_or_exception(
+        self,
+        klass: Union[Type[Model], QuerySet],
+        error_message: str = None,
+        exception: Type[APIException] = NotFound,
+        **kwargs: Any,
+    ) -> Any:
+        obj = get_object_or_exception(
+            klass=klass, error_message=error_message, exception=exception, **kwargs
+        )
+        self.check_object_permissions(obj)
+        return obj
+
+    def get_object_or_none(
+        self, klass: Union[Type[Model], QuerySet], **kwargs: Any
+    ) -> Optional[Any]:
+        obj = get_object_or_none(klass=klass, **kwargs)
+        if obj:
+            self.check_object_permissions(obj)
+        return obj
+
+    def _get_permissions(self) -> Iterable[BasePermission]:
         """
         Instantiates and returns the list of permissions that this view requires.
         """
@@ -197,7 +225,7 @@ def check_permissions(self) -> None:
         Check if the request should be permitted.
         Raises an appropriate exception if the request is not permitted.
         """
-        for permission in self.get_permissions():
+        for permission in self._get_permissions():
             if (
                 self.context
                 and self.context.request
@@ -207,12 +235,12 @@ def check_permissions(self) -> None:
             ):
                 self.permission_denied(permission)
 
-    def check_object_permissions(self, obj: Any) -> None:
+    def check_object_permissions(self, obj: Union[Any, Model]) -> None:
         """
         Check if the request should be permitted for a given object.
         Raises an appropriate exception if the request is not permitted.
         """
-        for permission in self.get_permissions():
+        for permission in self._get_permissions():
             if (
                 self.context
                 and self.context.request
@@ -222,7 +250,11 @@ def check_object_permissions(self, obj: Any) -> None:
             ):
                 self.permission_denied(permission)
 
-    def create_response(
-        self, message: Any, status_code: int = 200
-    ) -> ControllerResponse:
-        return self.Detail(message=message, status_code=status_code)
+    def create_response(self, message: Any, status_code: int = 200) -> HttpResponse:
+        response = self.Detail(message=message, status_code=status_code)
+        assert self.context and self.context.request and self.api
+        return self.api.create_response(
+            request=self.context.request,
+            data=response.convert_to_schema().dict(),
+            status=response.status_code,
+        )

ninja_extra/controllers/router.py

@@ -129,10 +129,10 @@ def urls_paths(self, prefix: str) -> Iterator[URLPattern]:
             # to skip lot of checks we simply treat double slash as a mistake:
             route = normalize_path(route)
             route = route.lstrip("/")
-
-            yield django_path(
-                route, path_view.get_view(), name=cast(str, path_view.url_name)
-            )
+            for op in path_view.operations:
+                yield django_path(
+                    route, path_view.get_view(), name=cast(str, op.url_name)
+                )
 
     def __repr__(self) -> str:
         return f"<controller - {self._controller.__name__}>"

tests/controllers.py

@@ -24,17 +24,27 @@ class EventSchemaOut(Schema):
 
 @router("events")
 class EventController(APIController):
-    @route.post(
-        "/create", url_name="event-create-url-name", response={201: EventSchemaOut}
-    )
+    @route.post("", url_name="event-create-url-name", response={201: EventSchemaOut})
     def create_event(self, event: EventSchema):
         event = Event.objects.create(**event.dict())
         return 201, event
 
-    @route.get("", response=List[EventSchema])
+    @route.get(
+        "",
+        response=List[EventSchema],
+        url_name="event-list",
+    )
     def list_events(self):
         return list(Event.objects.all())
 
+    @route.get(
+        "/list",
+        response=List[EventSchema],
+        url_name="event-list-2",
+    )
+    def list_events_example_2(self):
+        return list(Event.objects.all())
+
     @route.get("/{int:id}", response=EventSchema)
     def get_event(self, id: int):
         event = get_object_or_404(Event, id=id)

tests/test_controller.py

@@ -1,11 +1,10 @@
-from typing import Any, Type, Union
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import pytest
-from ninja import Schema
+from django.contrib.auth.models import Group
 
-from ninja_extra import APIController, NinjaExtraAPI, route, router
-from ninja_extra.controllers import RouteFunction
+from ninja_extra import APIController, NinjaExtraAPI, exceptions, route, router, testing
+from ninja_extra.controllers import RouteContext, RouteFunction
 from ninja_extra.controllers.base import MissingRouterDecoratorException
 from ninja_extra.controllers.response import Detail, Id, Ok
 from ninja_extra.controllers.router import ControllerRouter
@@ -98,6 +97,52 @@ def test_controller_route_definition_should_return_instance_route_definitions(se
         for route_definition in SomeControllerWithRoute.get_route_functions():
             assert isinstance(route_definition, RouteFunction)
 
+    @pytest.mark.django_db
+    def test_controller_get_object_or_exception_works(self):
+        group_instance = Group.objects.create(name="_groupowner")
+
+        controller_object = SomeController()
+        context = RouteContext(request=Mock(), permission_classes=[AllowAny])
+        controller_object.context = context
+        with patch.object(
+            AllowAny, "has_object_permission", return_value=True
+        ) as c_cop:
+            group = controller_object.get_object_or_exception(
+                Group, id=group_instance.id
+            )
+            c_cop.assert_called()
+            assert group == group_instance
+
+        with pytest.raises(Exception) as ex:
+            controller_object.get_object_or_exception(Group, id=1000)
+            assert isinstance(ex, exceptions.NotFound)
+
+        with pytest.raises(Exception) as ex:
+            with patch.object(AllowAny, "has_object_permission", return_value=False):
+                controller_object.get_object_or_exception(Group, id=group_instance.id)
+                assert isinstance(ex, exceptions.PermissionDenied)
+
+    @pytest.mark.django_db
+    def test_controller_get_object_or_none_works(self):
+        group_instance = Group.objects.create(name="_groupowner2")
+
+        controller_object = SomeController()
+        context = RouteContext(request=Mock(), permission_classes=[AllowAny])
+        controller_object.context = context
+        with patch.object(
+            AllowAny, "has_object_permission", return_value=True
+        ) as c_cop:
+            group = controller_object.get_object_or_none(Group, id=group_instance.id)
+            c_cop.assert_called()
+            assert group == group_instance
+
+        assert controller_object.get_object_or_none(Group, id=1000) is None
+
+        with pytest.raises(Exception) as ex:
+            with patch.object(AllowAny, "has_object_permission", return_value=False):
+                controller_object.get_object_or_none(Group, id=group_instance.id)
+                assert isinstance(ex, exceptions.PermissionDenied)
+
 
 class TestAPIControllerResponse:
     ok_response = Ok("OK")
@@ -122,10 +167,11 @@ def test_controller_response(self):
 
     def test_controller_response_works(self):
         detail = Detail("5242", status_code=302)
-        result = SomeControllerWithRouter.example2(request=Mock(), ex_id="5242")
-        assert isinstance(result, tuple)
-        assert result[1] == detail.convert_to_schema()
-        assert result[0] == detail.status_code
+        client = testing.TestClient(SomeControllerWithRouter)
+        response = client.get("/example/5242")
+
+        assert response.status_code == 302
+        assert detail.convert_to_schema().dict() == response.json()
 
         ok_response = Ok("5242")
         result = SomeControllerWithRouter.example_with_ok_response(

tests/test_django_model.py

@@ -11,7 +11,7 @@ def test_with_client(client: Client):
 
     test_item = {"start_date": "2020-01-01", "end_date": "2020-01-02", "title": "test"}
 
-    response = client.post("/api/events/create", **json_payload(test_item))
+    response = client.post("/api/events", **json_payload(test_item))
     assert response.status_code == 201
     assert Event.objects.count() == 1
 
@@ -26,7 +26,9 @@ def test_with_client(client: Client):
 
 def test_reverse():
     # check that url reversing works
-    assert reverse("api-1.0.0:event-create-url-name") == "/api/events/create"
+    assert reverse("api-1.0.0:event-create-url-name") == "/api/events"
+    assert reverse("api-1.0.0:event-list") == "/api/events"
+    assert reverse("api-1.0.0:event-list-2") == "/api/events/list"
 
 
 def json_payload(data):

tests/test_event_controller.py

@@ -20,7 +20,7 @@ class TestEventController:
     def test_create_event_works(self):
         client = TestClient(EventController)
         response = client.post(
-            "/create",
+            "",
             json=dict(
                 title="TestEvent1Title",
                 start_date=str(datetime.now().date()),