Create fastq analysis in trailblazer (#2833)(patch)

cg/cli/workflow/fastq/base.py

@@ -1,13 +1,12 @@
-import datetime as dt
 import logging
 
 import click
 
 from cg.cli.workflow.commands import ARGUMENT_CASE_ID
+from cg.cli.workflow.fastq.fastq_service import FastqService
 from cg.constants.constants import DRY_RUN, Pipeline
 from cg.meta.workflow.analysis import AnalysisAPI
 from cg.store import Store
-from cg.store.models import Analysis, Case
 
 LOG = logging.getLogger(__name__)
 
@@ -24,21 +23,16 @@ def fastq(context: click.Context):
 @ARGUMENT_CASE_ID
 @click.pass_context
 def store_fastq_analysis(context: click.Context, case_id: str, dry_run: bool = False):
-    """Creates an analysis object in status-db for the given fast case"""
     LOG.info(f"Creating an analysis for case {case_id}")
-    status_db: Store = context.obj.status_db
-    case_obj: Case = status_db.get_case_by_internal_id(internal_id=case_id)
-    new_analysis: Analysis = status_db.add_analysis(
-        pipeline=Pipeline.FASTQ,
-        completed_at=dt.datetime.now(),
-        primary=True,
-        started_at=dt.datetime.now(),
-        case_id=case_obj.id,
-    )
+
     if dry_run:
         return
-    status_db.session.add(new_analysis)
-    status_db.session.commit()
+
+    fastq_service = FastqService(
+        store=context.obj.status_db,
+        trailblazer_api=context.obj.trailblazer_api,
+    )
+    fastq_service.store_analysis(case_id)
 
 
 @fastq.command("store-available")

cg/cli/workflow/fastq/fastq_service.py

@@ -0,0 +1,49 @@
+import datetime as dt
+
+from cg.apps.tb.api import TrailblazerAPI
+from cg.constants.constants import AnalysisType, Pipeline
+from cg.constants.tb import AnalysisStatus
+from cg.exc import CaseNotFoundError
+from cg.store.api.core import Store
+from cg.store.models import Analysis, Case
+
+
+class FastqService:
+    def __init__(self, store: Store, trailblazer_api: TrailblazerAPI):
+        self.store = store
+        self.trailblazer_api = trailblazer_api
+
+    def store_analysis(self, case_id: str) -> None:
+        case: Case = self._get_case(case_id)
+        self._add_analysis_to_store(case)
+        self._add_analysis_to_trailblazer(case)
+
+    def _get_case(self, case_id: str) -> Case:
+        if case := self.store.get_case_by_internal_id(case_id):
+            return case
+        raise CaseNotFoundError(f"Case {case_id} not found.")
+
+    def _add_analysis_to_store(self, case: Case) -> None:
+        new_analysis: Analysis = self.store.add_analysis(
+            pipeline=Pipeline.FASTQ,
+            completed_at=dt.datetime.now(),
+            primary=True,
+            started_at=dt.datetime.now(),
+            case_id=case.id,
+        )
+        self.store.session.add(new_analysis)
+        self.store.session.commit()
+
+    def _add_analysis_to_trailblazer(self, case: Case) -> None:
+        self.trailblazer_api.add_pending_analysis(
+            case_id=case.internal_id,
+            analysis_type=AnalysisType.OTHER,
+            data_analysis=Pipeline.FASTQ,
+            config_path="",
+            out_dir="",
+            slurm_quality_of_service=case.slurm_priority,
+            ticket=case.latest_ticket,
+        )
+        self.trailblazer_api.set_analysis_status(
+            case_id=case.internal_id, status=AnalysisStatus.COMPLETED
+        )

cg/store/api/find_business_data.py

@@ -731,7 +731,7 @@ def is_case_external(self, case_id: str) -> bool:
         case: Case = self.get_case_by_internal_id(internal_id=case_id)
         return all(sample.application_version.application.is_external for sample in case.samples)
 
-    def get_case_by_internal_id(self, internal_id: str) -> Case:
+    def get_case_by_internal_id(self, internal_id: str) -> Case | None:
         """Get case by internal id."""
         return apply_case_filter(
             filter_functions=[CaseFilter.FILTER_BY_INTERNAL_ID],

cg/store/models.py

@@ -17,6 +17,7 @@
 )
 from cg.constants.archiving import PDC_ARCHIVE_LOCATION
 from cg.constants.constants import CONTROL_OPTIONS, CaseActions, PrepCategory
+from cg.constants.priority import SlurmQos
 
 Model = declarative_base()
 
@@ -487,6 +488,12 @@ def is_uploaded(self) -> bool:
         """Returns True if the latest connected analysis has been uploaded."""
         return self.analyses and self.analyses[0].uploaded_at
 
+    @property
+    def slurm_priority(self) -> SlurmQos:
+        case_priority: str = self.priority
+        slurm_priority: str = Priority.priority_to_slurm_qos().get(case_priority)
+        return SlurmQos(slurm_priority)
+
     def get_delivery_arguments(self) -> set[str]:
         """Translates the case data_delivery field to pipeline specific arguments."""
         delivery_arguments: set[str] = set()

tests/cli/workflow/fastq/test_fastq_base.py

@@ -9,17 +9,17 @@
 from cg.store.models import Analysis, Case, Sample
 
 
-def test_store_fastq_analysis(caplog, another_case_id: str, cli_runner, fastq_context, helpers):
+def test_store_fastq_analysis(another_case_id: str, cli_runner, fastq_context, helpers):
     """Test for CLI command creating an analysis object for a fastq case"""
     # GIVEN a fastq context
-    caplog.set_level(logging.INFO)
     helpers.ensure_case(fastq_context.status_db, case_id=another_case_id)
-    case_obj: Case = fastq_context.status_db.get_case_by_internal_id(internal_id=another_case_id)
+    case_obj: Case = fastq_context.status_db.get_case_by_internal_id(another_case_id)
     assert not case_obj.analyses
-    # WHEN the store_fastq_analysis command is invoked
+
+    # WHEN a command is run to create an analysis for the case
     cli_runner.invoke(store_fastq_analysis, [another_case_id], obj=fastq_context)
 
-    # THEN the run command should be reached
+    # THEN the analysis is created
     assert (
         len(
             fastq_context.status_db._get_query(table=Analysis)