Migrate to django-ninja

isic/conftest.py

@@ -40,9 +40,15 @@ def setup_groups(request):
 
 @pytest.fixture
 def api_client() -> APIClient:
+    # TODO have this return a django.test.Client instead of DRF's APIClient
     return APIClient()
 
 
+@pytest.fixture
+def client() -> Client:
+    return Client()
+
+
 @pytest.fixture
 def authenticated_client(user):
     # Do not use the client fixture, to prevent mutating its state

isic/core/api/__init__.py

@@ -1,9 +1,5 @@
-from .collection import CollectionViewSet
-from .image import ImageViewSet
 from .user import user_me
 
 __all__ = [
-    "CollectionViewSet",
-    "ImageViewSet",
     "user_me",
 ]

isic/core/api/collection.py

@@ -1,105 +1,126 @@
 from django.contrib import messages
-from django.contrib.auth.models import User
 from django.http.response import JsonResponse
-from django.utils.decorators import method_decorator
-from django_filters.rest_framework import DjangoFilterBackend
-from drf_yasg.utils import swagger_auto_schema
-from rest_framework import status
-from rest_framework.decorators import action
-from rest_framework.exceptions import PermissionDenied
-from rest_framework.response import Response
-from rest_framework.viewsets import ReadOnlyModelViewSet
+from django.shortcuts import get_object_or_404
+from ninja import Field, ModelSchema, Router, Schema
+from ninja.pagination import paginate
+from pydantic.types import conlist, constr
 
+from isic.core.constants import ISIC_ID_REGEX
 from isic.core.models.collection import Collection
-from isic.core.permissions import IsicObjectPermissionsFilter
-from isic.core.serializers import CollectionSerializer, IsicIdListSerializer, SearchQuerySerializer
+from isic.core.pagination import CursorPagination
+from isic.core.permissions import get_visible_objects
+from isic.core.serializers import SearchQueryIn
 from isic.core.services.collection.image import (
     collection_add_images_from_isic_ids,
     collection_remove_images_from_isic_ids,
 )
 from isic.core.tasks import populate_collection_from_search_task
 
-from .exceptions import Conflict
+router = Router()
 
 
-@method_decorator(
-    name="list", decorator=swagger_auto_schema(operation_summary="Return a list of collections.")
+class CollectionOut(ModelSchema):
+    class Config:
+        model = Collection
+        model_fields = ["id", "name", "description", "public", "pinned", "locked", "doi"]
+
+    doi_url: str | None = Field(alias="doi_url")
+
+
+@router.get("/", response=list[CollectionOut], summary="Return a list of collections.")
+@paginate(CursorPagination)
+def collection_list(request, pinned: bool | None = None) -> list[CollectionOut]:
+    queryset = get_visible_objects(request.user, "core.view_collection", Collection.objects.all())
+    if pinned is not None:
+        queryset = queryset.filter(pinned=pinned)
+    return queryset
+
+
+@router.get("/{id}/", response=CollectionOut, summary="Retrieve a single collection by ID.")
+def collection_detail(request, id: int) -> CollectionOut:
+    qs = get_visible_objects(request.user, "core.view_collection", Collection.objects.all())
+    collection = get_object_or_404(qs, id=id)
+    return collection
+
+
+@router.post(
+    "/{id}/populate-from-search/",
+    response={202: None, 403: dict, 409: dict},
+    include_in_schema=False,
+)
+def collection_populate_from_search(request, id: int, payload: SearchQueryIn):
+    qs = get_visible_objects(request.user, "core.view_collection", Collection.objects.all())
+    collection = get_object_or_404(qs, id=id)
+
+    if not request.user.has_perm("core.add_images", collection):
+        return 403, {"error": "You do not have permission to add images to this collection."}
+
+    if collection.locked:
+        return 409, {"error": "Collection is locked"}
+
+    if collection.public and payload.to_queryset(request.user).private().exists():
+        return 409, {"error": "Collection is public and cannot contain private images."}
+
+    # Pass data instead of validated_data because the celery task is going to revalidate.
+    # This avoids re encoding collections as a comma delimited string.
+    populate_collection_from_search_task.delay(id, request.user.pk, payload.dict())
+
+    # TODO: this is a weird mixture of concerns between SSR and an API, figure out a better
+    # way to handle this.
+    messages.add_message(
+        request, messages.INFO, "Adding images to collection, this may take a few minutes."
+    )
+    return 202, {}
+
+
+class IsicIdList(Schema):
+    isic_ids: conlist(constr(pattern=ISIC_ID_REGEX), max_length=500)
+
+
+# TODO: refactor *-from-list methods
+@router.post(
+    "/{id}/populate-from-list/", response={200: None, 403: dict, 409: dict}, include_in_schema=False
 )
-@method_decorator(
-    name="retrieve",
-    decorator=swagger_auto_schema(operation_summary="Retrieve a single collection by ID."),
+def collection_populate_from_list(request, id, payload: IsicIdList):
+    qs = get_visible_objects(request.user, "core.view_collection", Collection.objects.all())
+    collection = get_object_or_404(qs, id=id)
+
+    if not request.user.has_perm("core.add_images", collection):
+        return 403, {"error": "You do not have permission to add images to this collection."}
+
+    if collection.locked:
+        return 409, {"error": "Collection is locked"}
+
+    summary = collection_add_images_from_isic_ids(
+        user=request.user,
+        collection=collection,
+        isic_ids=payload.isic_ids,
+    )
+
+    return JsonResponse(summary)
+
+
+@router.post(
+    "/{id}/remove-from-list/", response={200: None, 403: dict, 409: dict}, include_in_schema=False
 )
-class CollectionViewSet(ReadOnlyModelViewSet):
-    serializer_class = CollectionSerializer
-    queryset = Collection.objects.all()
-    filter_backends = [IsicObjectPermissionsFilter, DjangoFilterBackend]
-    filterset_fields = ["pinned"]
-
-    def _enforce_write_checks(self, user: User):
-        if self.get_object().locked:
-            raise Conflict("Collection is locked for changes.")
-
-    @swagger_auto_schema(auto_schema=None)
-    @action(detail=True, methods=["post"], pagination_class=None, url_path="populate-from-search")
-    def populate_from_search(self, request, *args, **kwargs):
-        if not request.user.has_perm("core.add_images", self.get_object()):
-            raise PermissionDenied
-
-        self._enforce_write_checks(request.user)
-        serializer = SearchQuerySerializer(data=request.data, context={"user": request.user})
-        serializer.is_valid(raise_exception=True)
-
-        if self.get_object().public and serializer.to_queryset().private().exists():
-            raise Conflict("You are attempting to add private images to a public collection.")
-
-        # Pass data instead of validated_data because the celery task is going to revalidate.
-        # This avoids re encoding collections as a comma delimited string.
-        populate_collection_from_search_task.delay(kwargs["pk"], request.user.pk, serializer.data)
-
-        # TODO: this is a weird mixture of concerns between SSR and an API, figure out a better
-        # way to handle this.
-        messages.add_message(
-            request, messages.INFO, "Adding images to collection, this may take a few minutes."
-        )
-        return Response(status=status.HTTP_202_ACCEPTED)
-
-    # TODO: refactor *-from-list methods
-    @swagger_auto_schema(auto_schema=None)
-    @action(detail=True, methods=["post"], pagination_class=None, url_path="populate-from-list")
-    def populate_from_list(self, request, *args, **kwargs):
-        if not request.user.has_perm("core.add_images", self.get_object()):
-            raise PermissionDenied
-
-        self._enforce_write_checks(request.user)
-        serializer = IsicIdListSerializer(data=request.data)
-        serializer.is_valid(raise_exception=True)
-
-        summary = collection_add_images_from_isic_ids(
-            user=request.user,
-            collection=self.get_object(),
-            isic_ids=serializer.validated_data["isic_ids"],
-        )
-
-        return JsonResponse(summary)
-
-    @swagger_auto_schema(auto_schema=None)
-    @action(detail=True, methods=["post"], pagination_class=None, url_path="remove-from-list")
-    def remove_from_list(self, request, *args, **kwargs):
-        if not request.user.has_perm("core.remove_images", self.get_object()):
-            raise PermissionDenied
-
-        self._enforce_write_checks(request.user)
-        serializer = IsicIdListSerializer(data=request.data)
-        serializer.is_valid(raise_exception=True)
-
-        summary = collection_remove_images_from_isic_ids(
-            user=request.user,
-            collection=self.get_object(),
-            isic_ids=serializer.validated_data["isic_ids"],
-        )
-
-        # TODO: this is a weird mixture of concerns between SSR and an API, figure out a better
-        # way to handle this.
-        messages.add_message(request, messages.INFO, f'Removed {len(summary["succeeded"])} images.')
-
-        return JsonResponse(summary)
+def remove_from_list(request, id, payload: IsicIdList):
+    qs = get_visible_objects(request.user, "core.view_collection", Collection.objects.all())
+    collection = get_object_or_404(qs, id=id)
+
+    if not request.user.has_perm("core.remove_images", collection):
+        return 403, {"error": "You do not have permission to add images to this collection."}
+
+    if collection.locked:
+        return 409, {"error": "Collection is locked"}
+
+    summary = collection_remove_images_from_isic_ids(
+        user=request.user,
+        collection=collection,
+        isic_ids=payload.isic_ids,
+    )
+
+    # TODO: this is a weird mixture of concerns between SSR and an API, figure out a better
+    # way to handle this.
+    messages.add_message(request, messages.INFO, f'Removed {len(summary["succeeded"])} images.')
+
+    return JsonResponse(summary)