Lineage backend

backend/fms_core/router.py

@@ -26,6 +26,7 @@
     ProjectViewSet,
     SequenceViewSet,
     TaxonViewSet,
+    SampleLineageViewSet,
 )
 
 __all__ = ["router"]
@@ -56,3 +57,4 @@
 router.register(r"revisions", RevisionViewSet)
 router.register(r"users", UserViewSet)
 router.register(r"groups", GroupViewSet)
+router.register(r"sample-lineage", SampleLineageViewSet, basename="sample-lineage")

backend/fms_core/services/sample_lineage.py

@@ -1,5 +1,6 @@
+from django.db.models import F
 from django.core.exceptions import ValidationError
-from fms_core.models import SampleLineage
+from fms_core.models import SampleLineage, Sample, DerivedBySample, ProcessMeasurement
 
 
 def create_sample_lineage(parent_sample, child_sample, process_measurement):
@@ -25,3 +26,28 @@ def create_sample_lineage(parent_sample, child_sample, process_measurement):
             errors.append(str(e))
 
     return (sample_lineage, errors, warnings)
+
+def create_sample_lineage_graph(sampleId):
+    errors = []
+    nodes = []
+    edges = []
+
+    if not Sample.objects.filter(pk=sampleId).exists():
+        errors.append(f"Sample with id ${sampleId} does not exist")
+
+    if not errors:
+        derivedBySample = DerivedBySample.objects.filter(sample__id=sampleId) \
+                                                 .values_list("derived_sample__biosample_id", flat=True)
+
+        sampleIds = DerivedBySample.objects.filter(derived_sample__biosample__id__in=derivedBySample) \
+                                           .values_list("sample__id", flat=True)
+        samples = Sample.objects.filter(id__in=sampleIds)
+
+        process_measurements = ProcessMeasurement.objects.filter(source_sample__in=sampleIds) \
+                                                         .select_related("process__lineage").annotate(child_sample=F("lineage__child")) \
+                                                         .select_related("process_protocol").annotate(protocol_name=F("process__protocol__name"))
+
+        nodes = list(samples.values("id", "name", "quality_flag", "quantity_flag"))
+        edges = list(process_measurements.values("id", "source_sample", "child_sample", "protocol_name"))
+
+    return (nodes, edges, errors)

backend/fms_core/viewsets/__init__.py

@@ -23,6 +23,7 @@
 from .platform import PlatformViewSet
 from .library_type import LibraryTypeViewSet
 from .taxon import TaxonViewSet
+from .sample_lineage import SampleLineageViewSet
 
 __all__ = [
     "ContainerViewSet",
@@ -50,4 +51,5 @@
     "PlatformViewSet",
     "LibraryTypeViewSet",
     "TaxonViewSet",
+    "SampleLineageViewSet"
 ]

backend/fms_core/viewsets/sample_lineage.py

@@ -0,0 +1,21 @@
+from fms_core.services.sample_lineage import create_sample_lineage_graph
+from rest_framework import viewsets
+from rest_framework.decorators import action
+from rest_framework.response import Response
+from django.core.exceptions import ValidationError
+
+from fms_core.models import Sample
+from fms_core.serializers import SampleSerializer, ProcessMeasurementSerializer
+
+class SampleLineageViewSet(viewsets.ViewSet):
+    @action(detail=True, methods=["get"])
+    def graph(self, _request, pk):
+        nodes, edges, errors = create_sample_lineage_graph(pk)
+
+        if errors:
+            raise ValidationError(errors)
+        else:
+            return Response({
+                "nodes": nodes,
+                "edges": edges,
+            })