Refactored with tooling from libumccr 0.3.0

data_portal/viewsets/labmetadata.py

@@ -6,10 +6,9 @@
 """
 import logging
 import os
-from enum import Enum
 
-import boto3
-from libumccr import libjson
+from libumccr import libjson, aws
+from libumccr.aws.liblambda import LambdaInvocationType
 from rest_framework import filters, status
 from rest_framework.decorators import action
 from rest_framework.response import Response
@@ -22,12 +21,6 @@
 logger = logging.getLogger(__name__)
 
 
-class LambdaInvocationType(Enum):
-    EVENT = 'Event'
-    REQUEST_RESPONSE = 'RequestResponse'
-    DRY_RUN = 'DryRun'
-
-
 class LabMetadataViewSet(ReadOnlyModelViewSet):
     serializer_class = LabMetadataModelSerializer
     pagination_class = StandardResultsSetPagination
@@ -50,7 +43,7 @@ def sync(self, request):
 
             logger.debug(payload_json)
 
-            client = boto3.client('lambda')
+            client = aws.lambda_client()
             lmbda_response = client.invoke(
                 FunctionName=fn_name,
                 InvocationType=LambdaInvocationType.EVENT.value,

data_processors/pipeline/domain/manops.py

@@ -8,28 +8,16 @@
 """
 import json
 import os
-from typing import List
 from abc import ABC, abstractmethod
 from enum import Enum
+from typing import List
+
 from libumccr import aws
+from libumccr.aws.liblambda import LambdaInvocationType
+
 from data_portal.models.workflow import Workflow
 from data_processors.pipeline.domain.workflow import WorkflowType, WorkflowStatus
-from data_processors.pipeline.services.workflow_srv import get_workflows_by_subject_id_and_workflow_type, \
-    get_labmetadata_by_wfr_id
-import boto3
-
-
-############################################################
-# Improvement: importing from libumccr when it is available
-# Ref: https://github.com/umccr/libumccr/commit/cbf781fbd271e63c17dd45bd73751bf4807c1f6c
-
-class LambdaInvocationType(Enum):
-    EVENT = 'Event'
-    REQUEST_RESPONSE = 'RequestResponse'
-    DRY_RUN = 'DryRun'
-
-
-############################################################
+from data_processors.pipeline.services import workflow_srv
 
 
 class Report(Enum):
@@ -69,17 +57,18 @@ def add_dataset(self, dataset: str):
     def add_workflow(self, wfr_id: str):
         self.wfr_id = wfr_id
 
-        matching_labmetadata = get_labmetadata_by_wfr_id(wfr_id=wfr_id)
+        matching_labmetadata = workflow_srv.get_labmetadata_by_wfr_id(wfr_id=wfr_id)
 
         # Get subject_id, it should only contain one labmetadata
         self.subject_id = matching_labmetadata[0].subject_id
 
     def add_workflow_from_subject(self, subject_id: str):
         self.subject_id = subject_id
 
-        workflow_list: List[Workflow] = get_workflows_by_subject_id_and_workflow_type(subject_id=subject_id,
-                                                                                      workflow_type=WorkflowType.UMCCRISE
-                                                                                      )
+        workflow_list: List[Workflow] = workflow_srv.get_workflows_by_subject_id_and_workflow_type(
+            subject_id=subject_id,
+            workflow_type=WorkflowType.UMCCRISE,
+        )
         # Set if value exist
         if len(workflow_list) > 0:
             self.wfr_id = workflow_list[0].wfr_id
@@ -88,10 +77,11 @@ def generate(self):
         fn_name = os.getenv('MANOPS_LAMBDA', 'data-portal-api-dev-manops')
 
         # Check if existing RNAsum workflows is running
-        workflow_list: List[Workflow] = get_workflows_by_subject_id_and_workflow_type(subject_id=self.subject_id,
-                                                                                      workflow_type=WorkflowType.RNASUM,
-                                                                                      workflow_status=WorkflowStatus.RUNNING
-                                                                                      )
+        workflow_list: List[Workflow] = workflow_srv.get_workflows_by_subject_id_and_workflow_type(
+            subject_id=self.subject_id,
+            workflow_type=WorkflowType.RNASUM,
+            workflow_status=WorkflowStatus.RUNNING,
+        )
 
         if len(workflow_list) > 0:
             # Current RNAsum workflow has run. Terminating
@@ -103,7 +93,7 @@ def generate(self):
             "dataset": self.dataset
         })
 
-        lambda_client = boto3.client('lambda')
+        lambda_client = aws.lambda_client()
         lambda_response = lambda_client.invoke(
             FunctionName=fn_name,
             InvocationType=LambdaInvocationType.EVENT.value,

data_processors/pipeline/domain/somalier.py

@@ -22,14 +22,6 @@
 logger.setLevel(logging.INFO)
 
 
-def stepfn_client(**kwargs):
-    return aws.client('stepfunctions', **kwargs)
-
-
-def srv_discovery_client(**kwargs):
-    return aws.client('servicediscovery', **kwargs)
-
-
 class SomalierInterface(ABC):
     """Somalier Interface Contract
 
@@ -103,8 +95,8 @@ class HolmesPipeline(HolmesInterface):
     EXTRACT_STEPS_ARN_KEY = "extractStepsArn"
 
     def __init__(self):
-        self.srv_discovery_client = srv_discovery_client()
-        self.stepfn_client = stepfn_client()
+        self.srv_discovery_client = aws.srv_discovery_client()
+        self.stepfn_client = aws.stepfn_client()
 
         self.service_id = self.__discover_service_id()
         self.service_attributes = self.__discover_service_attributes()

data_processors/pipeline/lambdas/fastq_list_row.py

@@ -15,8 +15,7 @@
 from typing import List
 import pandas as pd
 
-from data_processors.pipeline.tools import libregex
-from libumccr import libjson
+from libumccr import libjson, libregex
 from libica.app import gds
 
 logger = logging.getLogger()

data_processors/pipeline/orchestration/google_lims_update_step.py

@@ -7,14 +7,14 @@
 from typing import List, Set
 
 from django.db.models import QuerySet
-from libumccr import libgdrive
+from libumccr import libgdrive, libregex
 from libumccr.aws import libssm
 
 from data_portal.models.labmetadata import LabMetadata
 from data_portal.models.limsrow import LIMSRow
 from data_portal.models.workflow import Workflow
 from data_processors import const
-from data_processors.pipeline.tools import liborca, libregex
+from data_processors.pipeline.tools import liborca
 
 
 def perform(workflow: Workflow):

data_processors/pipeline/services/libraryrun_srv.py

@@ -5,11 +5,12 @@
 
 from django.db import transaction
 from django.db.models import QuerySet
+from libumccr import libregex
 
 from data_portal.models.libraryrun import LibraryRun
 from data_portal.models.workflow import Workflow
 from data_processors.pipeline.services import metadata_srv
-from data_processors.pipeline.tools import liborca, libregex
+from data_processors.pipeline.tools import liborca
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)

data_processors/pipeline/tools/liborca.py

@@ -21,11 +21,9 @@
 from typing import List, Dict, Any
 
 from libica.app import gds
-from libumccr import libjson
+from libumccr import libjson, libregex
 from sample_sheet import SampleSheet
 
-from data_processors.pipeline.tools import libregex
-
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
 

requirements.txt

@@ -10,7 +10,7 @@ python-dateutil==2.8.2
 google-auth==2.9.1
 Werkzeug==2.2.0
 libica==2.1.0
-libumccr==0.2.0
+libumccr==0.3.0
 gspread==5.4.0
 gspread-pandas==3.2.2
 sample-sheet==0.12.0