refactor assessment permissions in django views and API

hawc/apps/animal/api.py

@@ -14,45 +14,37 @@
     get_assessment_from_query,
     get_assessment_id_param,
 )
-from ..assessment.models import Assessment
-from ..common.api import (
-    CleanupFieldsBaseViewSet,
-    LegacyAssessmentAdapterMixin,
-    user_can_edit_object,
-)
+from ..common.api import CleanupFieldsBaseViewSet, user_can_edit_object
 from ..common.helper import FlatExport, re_digits
 from ..common.renderers import PandasRenderers
 from ..common.serializers import HeatmapQuerySerializer, UnusedSerializer
-from ..common.views import AssessmentPermissionsMixin, create_object_log
+from ..common.views import create_object_log
 from . import exports, models, serializers
 from .actions.model_metadata import AnimalMetadata
 from .actions.term_check import term_check
 
 
-class AnimalAssessmentViewset(
-    AssessmentPermissionsMixin, LegacyAssessmentAdapterMixin, viewsets.GenericViewSet
-):
-    parent_model = Assessment
-    model = models.Endpoint
+class AnimalAssessmentViewset(viewsets.GenericViewSet):
+    model = models.Assessment
+    queryset = models.Assessment.objects.all()
     permission_classes = (AssessmentLevelPermissions,)
     serializer_class = UnusedSerializer
     lookup_value_regex = re_digits
 
-    def get_queryset(self):
-        perms = self.get_obj_perms()
+    def get_endpoint_queryset(self):
+        perms = self.assessment.get_obj_perms()
         if not perms["edit"]:
-            return self.model.objects.published(self.assessment)
-        return self.model.objects.get_qs(self.assessment)
+            return models.Endpoint.objects.published(self.assessment)
+        return models.Endpoint.objects.get_qs(self.assessment)
 
     @action(detail=True, url_path="full-export", renderer_classes=PandasRenderers)
     def full_export(self, request, pk):
         """
         Retrieve complete animal data
         """
-        self.set_legacy_attr(pk)
-        self.permission_check_user_can_view()
+        self.assessment = self.get_object()
         exporter = exports.EndpointGroupFlatComplete(
-            self.get_queryset(),
+            self.get_endpoint_queryset(),
             filename=f"{self.assessment}-bioassay-complete",
             assessment=self.assessment,
         )
@@ -63,10 +55,9 @@ def endpoint_export(self, request, pk):
         """
         Retrieve endpoint animal data
         """
-        self.set_legacy_attr(pk)
-        self.permission_check_user_can_view()
+        self.assessment = self.get_object()
         exporter = exports.EndpointSummary(
-            self.get_queryset(),
+            self.get_endpoint_queryset(),
             filename=f"{self.assessment}-bioassay-summary",
             assessment=self.assessment,
         )
@@ -80,8 +71,7 @@ def study_heatmap(self, request, pk):
         By default only shows data from published studies. If the query param `unpublished=true`
         is present then results from all studies are shown.
         """
-        self.set_legacy_attr(pk)
-        self.permission_check_user_can_view()
+        self.assessment = self.get_object()
         ser = HeatmapQuerySerializer(data=request.query_params)
         ser.is_valid(raise_exception=True)
         unpublished = ser.data["unpublished"]
@@ -103,8 +93,7 @@ def endpoint_heatmap(self, request, pk):
         By default only shows data from published studies. If the query param `unpublished=true`
         is present then results from all studies are shown.
         """
-        self.set_legacy_attr(pk)
-        self.permission_check_user_can_view()
+        self.assessment = self.get_object()
         ser = HeatmapQuerySerializer(data=request.query_params)
         ser.is_valid(raise_exception=True)
         unpublished = ser.data["unpublished"]
@@ -126,8 +115,7 @@ def endpoint_doses_heatmap(self, request, pk):
         By default only shows data from published studies. If the query param `unpublished=true`
         is present then results from all studies are shown.
         """
-        self.set_legacy_attr(pk)
-        self.permission_check_user_can_view()
+        self.assessment = self.get_object()
         ser = HeatmapQuerySerializer(data=request.query_params)
         ser.is_valid(raise_exception=True)
         unpublished = ser.data["unpublished"]
@@ -143,8 +131,7 @@ def endpoint_doses_heatmap(self, request, pk):
 
     @action(detail=True, renderer_classes=PandasRenderers)
     def endpoints(self, request, pk):
-        self.set_legacy_attr(pk)
-        self.permission_check_user_can_view()
+        self.assessment = self.get_object()
         ser = HeatmapQuerySerializer(data=request.query_params)
         ser.is_valid(raise_exception=True)
         unpublished = ser.data["unpublished"]
@@ -162,8 +149,7 @@ def endpoints(self, request, pk):
 
     @action(detail=True, url_path="ehv-check", renderer_classes=PandasRenderers)
     def ehv_check(self, request, pk):
-        self.set_legacy_attr(pk)
-        self.permission_check_user_can_edit()
+        _ = self.get_object()
         df = term_check(pk)
         export = FlatExport(df, f"term-report-{pk}")
         return Response(export)
@@ -184,7 +170,7 @@ def get_queryset(self):
 
     @transaction.atomic
     def perform_create(self, serializer):
-        # permissions check
+        # permissions check # todo herE?
         user_can_edit_object(serializer.study, self.request.user, raise_exception=True)
         super().perform_create(serializer)
         create_object_log(

hawc/apps/assessment/api/__init__.py

@@ -0,0 +1,5 @@
+from .filters import *  # noqa: F401,F403
+from .helper import *  # noqa: F401,F403
+from .pagination import *  # noqa: F401,F403
+from .permissions import *  # noqa: F401,F403
+from .viewsets import *  # noqa: F401,F403

hawc/apps/assessment/api/filters.py

@@ -0,0 +1,25 @@
+from rest_framework import filters
+
+from .helper import get_assessment_from_query
+
+
+class InAssessmentFilter(filters.BaseFilterBackend):
+    """
+    Filter objects which are in a particular assessment.
+    """
+
+    default_list_actions = ["list"]
+
+    def filter_queryset(self, request, queryset, view):
+        list_actions = getattr(view, "list_actions", self.default_list_actions)
+        if view.action not in list_actions:
+            return queryset
+
+        if not hasattr(view, "assessment"):
+            view.assessment = get_assessment_from_query(request)
+
+        if not view.assessment_filter_args:
+            raise ValueError("Viewset requires the `assessment_filter_args` argument")
+
+        filters = {view.assessment_filter_args: view.assessment.id}
+        return queryset.filter(**filters)

hawc/apps/assessment/api/helper.py

@@ -0,0 +1,36 @@
+from typing import Optional
+
+from rest_framework import status
+from rest_framework.exceptions import APIException
+
+from ...common.helper import tryParseInt
+from .. import models
+
+
+class RequiresAssessmentID(APIException):
+    status_code = status.HTTP_400_BAD_REQUEST
+    default_detail = "Please provide an `assessment_id` argument to your GET request."
+
+
+class InvalidAssessmentID(APIException):
+    status_code = status.HTTP_400_BAD_REQUEST
+    default_detail = "Assessment does not exist for given `assessment_id`."
+
+
+def get_assessment_id_param(request) -> int:
+    """
+    If request doesn't contain an integer-based `assessment_id`, an exception is raised.
+    """
+    assessment_id = tryParseInt(request.GET.get("assessment_id"))
+    if assessment_id is None:
+        raise RequiresAssessmentID()
+    return assessment_id
+
+
+def get_assessment_from_query(request) -> Optional[models.Assessment]:
+    """Returns assessment or raises exception if does not exist."""
+    assessment_id = get_assessment_id_param(request)
+    try:
+        return models.Assessment.objects.get(pk=assessment_id)
+    except models.Assessment.DoesNotExist:
+        raise InvalidAssessmentID()

hawc/apps/assessment/api/pagination.py

@@ -0,0 +1,5 @@
+from rest_framework.pagination import PageNumberPagination
+
+
+class DisabledPagination(PageNumberPagination):
+    page_size = None

hawc/apps/assessment/api/permissions.py

@@ -0,0 +1,33 @@
+from rest_framework import permissions
+
+
+class JobPermissions(permissions.BasePermission):
+    """
+    Requires admin permissions where jobs have no associated assessment
+    or when part of a list, and assessment level permissions when jobs
+    have an associated assessment.
+    """
+
+    def has_object_permission(self, request, view, obj):
+        if obj.assessment is None:
+            return bool(request.user and request.user.is_staff)
+        elif request.method in permissions.SAFE_METHODS:
+            return obj.assessment.user_can_view_object(request.user)
+        else:
+            return obj.assessment.user_can_edit_object(request.user)
+
+    def has_permission(self, request, view):
+        if view.action == "list":
+            return bool(request.user and request.user.is_staff)
+        elif view.action == "create":
+            serializer = view.serializer_class(data=request.data)
+            serializer.is_valid(raise_exception=True)
+            assessment = serializer.validated_data.get("assessment")
+            if assessment is None:
+                return bool(request.user and request.user.is_staff)
+            else:
+                return assessment.user_can_edit_object(request.user)
+        else:
+            # other actions are object specific,
+            # and will be caught by object permissions
+            return True