Feat(MM): Additional MM Endpoint and Library Filter (#765)

lib/workload/stateless/stacks/metadata-manager/app/tests/test_viewsets.py

@@ -1,13 +1,18 @@
+import json
 import logging
 
 from django.test import TestCase
 
+from app.models import Library, Sample
 from app.tests.factories import LIBRARY_1, SUBJECT_1, SAMPLE_1
-from app.tests.utils import insert_mock_1
+from app.tests.utils import insert_mock_1, is_obj_exists
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
+# pragma: allowlist nextline secret
+TEST_JWT = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyLCJlbWFpbCI6ImpvaG4uZG9lQGV4YW1wbGUuY29tIn0.1XOO35Ozn1XNEj_W7RFefNfJnVm7C1pm7MCEBPbCkJ4"
+
 
 def version_endpoint(ep: str):
     return "api/v1/" + ep
@@ -68,3 +73,32 @@ def test_get_api(self):
                 "No results are expected for unrecognized query parameter",
             )
 
+    def test_delete_api(self):
+        """
+        python manage.py test app.tests.test_viewsets.LabViewSetTestCase.test_delete_api
+        """
+
+        library = Library.objects.get(library_id=LIBRARY_1['library_id'])
+        self.client.delete(f"/{version_endpoint(f"library/{library.orcabus_id}/")}",
+                           headers={'Authorization': f'Bearer {TEST_JWT}'})
+        self.assertFalse(is_obj_exists(Library, library_id=LIBRARY_1['library_id']), "Library should be deleted")
+
+        sample = Sample.objects.get(sample_id=SAMPLE_1['sample_id'])
+        self.client.delete(f"/{version_endpoint(f"sample/{sample.orcabus_id}/")}",
+                           headers={'Authorization': f'Bearer {TEST_JWT}'})
+        self.assertFalse(is_obj_exists(Sample, sample_id=SAMPLE_1['sample_id']), "Sample should be deleted")
+
+    def test_patch_api(self):
+        """
+        python manage.py test app.tests.test_viewsets.LabViewSetTestCase.test_patch_api
+        """
+        new_coverage = 10.0
+
+        library = Library.objects.get(library_id=LIBRARY_1['library_id'])
+
+        self.assertEqual(library.coverage, LIBRARY_1['coverage'], "Coverage should be the same")
+        self.client.patch(f"/{version_endpoint(f"library/{library.orcabus_id}/")}",
+                                data=json.dumps({"coverage":new_coverage}),
+                                headers={'Authorization': f'Bearer {TEST_JWT}', 'Content-Type': 'application/json'})
+        library = Library.objects.get(library_id=LIBRARY_1['library_id'])
+        self.assertEqual(library.coverage, new_coverage, "Coverage should be updated")

lib/workload/stateless/stacks/metadata-manager/app/tests/utils.py

@@ -1,3 +1,5 @@
+from django.core.exceptions import ObjectDoesNotExist
+
 from app.models import Subject, Sample, Library, Project, Contact, Individual
 from app.tests.factories import LibraryFactory, IndividualFactory, SubjectFactory, SampleFactory, \
     ProjectFactory, ContactFactory
@@ -35,3 +37,11 @@ def insert_mock_1():
 
     subject.individual_set.add(individual)
     subject.save()
+
+
+def is_obj_exists(obj, **kwargs):
+    try:
+        obj.objects.get(**kwargs)
+        return True
+    except ObjectDoesNotExist:
+        return False

lib/workload/stateless/stacks/metadata-manager/app/viewsets/base.py

@@ -1,21 +1,27 @@
 from abc import ABC
 
+from drf_spectacular.utils import extend_schema
+from rest_framework.mixins import DestroyModelMixin
+
 from app.pagination import StandardResultsSetPagination
 
 from django.shortcuts import get_object_or_404
 
-from rest_framework import filters
+from rest_framework import filters, status
 from rest_framework.response import Response
-from rest_framework.viewsets import ReadOnlyModelViewSet
+from rest_framework.viewsets import ReadOnlyModelViewSet, ModelViewSet
+
+from app.viewsets.utils import get_email_from_jwt
 
 
-class BaseViewSet(ReadOnlyModelViewSet, ABC):
+class BaseViewSet(ModelViewSet, ABC):
     lookup_value_regex = "[^/]+"  # This is to allow for special characters in the URL
     orcabus_id_prefix = ''
     ordering_fields = "__all__"
     ordering = ["-orcabus_id"]
     pagination_class = StandardResultsSetPagination
     filter_backends = [filters.OrderingFilter, filters.SearchFilter]
+    http_method_names = ['get', 'patch', 'delete']
 
     def retrieve(self, request, *args, **kwargs):
         """
@@ -75,3 +81,38 @@ def retrieve_history(self, request, *args, **kwargs):
         serializer = history_serializer(page, many=True)
 
         return self.get_paginated_response(serializer.data)
+
+
+    def perform_destroy(self, instance):
+        """
+        The perform_destroy method is overridden to allow for the _history_user to be set.
+        """
+        requester_email = get_email_from_jwt(self.request)
+        if not requester_email:
+            raise ValueError("The requester email is not found in the JWT token.")
+
+        instance._history_user = requester_email
+        super().perform_destroy(instance)
+
+
+    def perform_update(self, serializer):
+        """
+        The perform_destroy method is overridden to allow for the _history_user to be set.
+        """
+        requester_email = get_email_from_jwt(self.request)
+        if not requester_email:
+            raise ValueError("The requester email is not found in the JWT token.")
+
+        serializer._history_user = requester_email
+        super().perform_update(serializer)
+
+    def perform_create(self, serializer):
+        """
+        The perform_create method is overridden to allow for the _history_user to be set.
+        """
+        requester_email = get_email_from_jwt(self.request)
+        if not requester_email:
+            raise ValueError("The requester email is not found in the JWT token.")
+
+        serializer._history_user = requester_email
+        super().perform_create(serializer)

lib/workload/stateless/stacks/metadata-manager/app/viewsets/contact.py

@@ -8,20 +8,32 @@
 
 
 class ContactViewSet(BaseViewSet):
-    serializer_class = ContactDetailSerializer
+    serializer_class = ContactSerializer
     search_fields = Contact.get_base_fields()
-    queryset = Contact.objects.prefetch_related('project_set').all()
+    queryset = Contact.objects.all()
     orcabus_id_prefix = Contact.orcabus_id_prefix
 
-    @extend_schema(parameters=[
-        ContactSerializer
-    ])
+    def get_queryset(self):
+        query_params = super().get_query_params()
+        return Contact.objects.get_by_keyword(**query_params)
+
+    @extend_schema(responses=ContactDetailSerializer(many=False))
+    def retrieve(self, request, *args, **kwargs):
+        self.serializer_class = ContactDetailSerializer
+        self.queryset = Contact.objects.prefetch_related('project_set').all()
+        return super().retrieve(request, *args, **kwargs)
+
+    @extend_schema(
+        parameters=[
+            ContactSerializer
+        ],
+        responses=ContactDetailSerializer(many=True),
+    )
     def list(self, request, *args, **kwargs):
+        self.serializer_class = ContactDetailSerializer
+        self.queryset = Contact.objects.prefetch_related('project_set').all()
         return super().list(request, *args, **kwargs)
 
-    def get_queryset(self):
-        query_params = self.get_query_params()
-        return Contact.objects.get_by_keyword(**query_params)
 
     @extend_schema(responses=ContactHistorySerializer(many=True), description="Retrieve the history of this model")
     @action(detail=True, methods=['get'], url_name='history', url_path='history')

lib/workload/stateless/stacks/metadata-manager/app/viewsets/individual.py

@@ -2,21 +2,30 @@
 from rest_framework.decorators import action
 
 from app.models import Individual
-from app.serializers.individual import IndividualDetailSerializer, IndividualHistorySerializer
+from app.serializers.individual import IndividualDetailSerializer, IndividualHistorySerializer, IndividualSerializer
 
 from .base import BaseViewSet
 
 
 class IndividualViewSet(BaseViewSet):
-    serializer_class = IndividualDetailSerializer
+    serializer_class = IndividualSerializer
     search_fields = Individual.get_base_fields()
-    queryset = Individual.objects.prefetch_related('subject_set').all()
+    queryset = Individual.objects.all()
     orcabus_id_prefix = Individual.orcabus_id_prefix
 
-    @extend_schema(parameters=[
-        IndividualDetailSerializer
-    ])
+    @extend_schema(responses=IndividualDetailSerializer(many=False))
+    def retrieve(self, request, *args, **kwargs):
+        self.serializer_class = IndividualDetailSerializer
+        self.queryset = Individual.objects.prefetch_related('subject_set').all()
+        return super().retrieve(request, *args, **kwargs)
+
+    @extend_schema(
+        parameters=[IndividualDetailSerializer],
+        responses=IndividualDetailSerializer(many=True),
+    )
     def list(self, request, *args, **kwargs):
+        self.serializer_class = IndividualDetailSerializer
+        self.queryset = Individual.objects.prefetch_related('subject_set').all()
         return super().list(request, *args, **kwargs)
 
     def get_queryset(self):

lib/workload/stateless/stacks/metadata-manager/app/viewsets/library.py

@@ -8,9 +8,10 @@
 
 
 class LibraryViewSet(BaseViewSet):
-    serializer_class = LibraryDetailSerializer
+    serializer_class = LibrarySerializer
+    detail_serializer_class = LibraryDetailSerializer
     search_fields = Library.get_base_fields()
-    queryset = Library.objects.select_related('sample').select_related('subject').prefetch_related('project_set').all()
+    queryset = Library.objects.all()
     orcabus_id_prefix = Library.orcabus_id_prefix
 
     def get_queryset(self):
@@ -35,22 +36,35 @@ def get_queryset(self):
         # Continue filtering by the keys inside the library model
         return Library.objects.get_by_keyword(qs, **query_params)
 
-    @extend_schema(parameters=[
-        LibrarySerializer,
-        OpenApiParameter(name='coverage[lte]',
-                         description="Filter based on 'coverage' that is less than or equal to the given value.",
-                         required=False,
-                         type=float),
-        OpenApiParameter(name='coverage[gte]',
-                         description="Filter based on 'coverage' that is greater than or equal to the given value.",
-                         required=False,
-                         type=float),
-        OpenApiParameter(name='project_id',
-                         description="Filter where the associated the project has the given 'project_id'.",
-                         required=False,
-                         type=float),
-    ])
+    @extend_schema(responses=LibraryDetailSerializer(many=False))
+    def retrieve(self, request, *args, **kwargs):
+        self.serializer_class = LibraryDetailSerializer
+        self.queryset = Library.objects.select_related('sample').select_related('subject').prefetch_related(
+            'project_set').all()
+        return super().retrieve(request, *args, **kwargs)
+
+    @extend_schema(
+        parameters=[
+            LibrarySerializer,
+            OpenApiParameter(name='coverage[lte]',
+                             description="Filter based on 'coverage' that is less than or equal to the given value.",
+                             required=False,
+                             type=float),
+            OpenApiParameter(name='coverage[gte]',
+                             description="Filter based on 'coverage' that is greater than or equal to the given value.",
+                             required=False,
+                             type=float),
+            OpenApiParameter(name='project_id',
+                             description="Filter where the associated the project has the given 'project_id'.",
+                             required=False,
+                             type=float),
+        ],
+        responses=LibraryDetailSerializer(many=True),
+    )
     def list(self, request, *args, **kwargs):
+        self.serializer_class = LibraryDetailSerializer
+        self.queryset = Library.objects.select_related('sample').select_related('subject').prefetch_related(
+            'project_set').all()
         return super().list(request, *args, **kwargs)
 
     @extend_schema(responses=LibraryHistorySerializer(many=True), description="Retrieve the history of this model")

lib/workload/stateless/stacks/metadata-manager/app/viewsets/project.py

@@ -8,15 +8,26 @@
 
 
 class ProjectViewSet(BaseViewSet):
-    serializer_class = ProjectDetailSerializer
+    serializer_class = ProjectSerializer
     search_fields = Project.get_base_fields()
-    queryset = Project.objects.prefetch_related("contact_set").all()
+    queryset = Project.objects.all()
     orcabus_id_prefix = Project.orcabus_id_prefix
 
-    @extend_schema(parameters=[
-        ProjectSerializer
-    ])
+    @extend_schema(responses=ProjectDetailSerializer(many=False))
+    def retrieve(self, request, *args, **kwargs):
+        self.serializer_class = ProjectDetailSerializer
+        self.queryset = Project.objects.prefetch_related("contact_set").all()
+        return super().retrieve(request, *args, **kwargs)
+
+    @extend_schema(
+        parameters=[
+            ProjectSerializer
+        ],
+        responses=ProjectDetailSerializer(many=True),
+    )
     def list(self, request, *args, **kwargs):
+        self.serializer_class = ProjectDetailSerializer
+        self.queryset = Project.objects.prefetch_related("contact_set").all()
         return super().list(request, *args, **kwargs)
 
     def get_queryset(self):

lib/workload/stateless/stacks/metadata-manager/app/viewsets/sample.py

@@ -1,4 +1,4 @@
-from drf_spectacular.utils import extend_schema
+from drf_spectacular.utils import extend_schema, OpenApiParameter
 from rest_framework.decorators import action
 
 from app.models import Sample
@@ -8,20 +8,43 @@
 
 
 class SampleViewSet(BaseViewSet):
-    serializer_class = SampleDetailSerializer
+    serializer_class = SampleSerializer
     search_fields = Sample.get_base_fields()
     queryset = Sample.objects.all()
     orcabus_id_prefix = Sample.orcabus_id_prefix
 
-    @extend_schema(parameters=[
-        SampleSerializer
-    ])
-    def list(self, request, *args, **kwargs):
-        return super().list(request, *args, **kwargs)
-
     def get_queryset(self):
+        qs = self.queryset
         query_params = self.get_query_params()
-        return Sample.objects.get_by_keyword(**query_params)
+
+        is_library_none = query_params.getlist("is_library_none", None)
+        if is_library_none:
+            query_params.pop("is_library_none")
+            qs = qs.filter(library=None)
+
+        return Sample.objects.get_by_keyword(qs, **query_params)
+
+    @extend_schema(responses=SampleDetailSerializer(many=False))
+    def retrieve(self, request, *args, **kwargs):
+        self.serializer_class = SampleDetailSerializer
+        self.queryset = Sample.objects.prefetch_related('library_set').all()
+        return super().retrieve(request, *args, **kwargs)
+
+
+    @extend_schema(
+        parameters=[
+            SampleSerializer,
+            OpenApiParameter(name='is_library_none',
+                             description="Filter where it is not linked to a library.",
+                             required=False,
+                             type=bool),
+        ],
+        responses=SampleDetailSerializer(many=True),
+    )
+    def list(self, request, *args, **kwargs):
+        self.queryset = Sample.objects.prefetch_related('library_set').all()
+        self.serializer_class = SampleDetailSerializer
+        return super().list(request, *args, **kwargs)
 
     @extend_schema(responses=SampleHistorySerializer(many=True), description="Retrieve the history of this model")
     @action(detail=True, methods=['get'], url_name='history', url_path='history')