add discriminator to nexflow qc models

cg/meta/workflow/nf_analysis.py

@@ -930,20 +930,25 @@ def get_gene_panel(self, case_id: str, dry_run: bool = False) -> list[str]:
             dry_run=dry_run,
         )
 
-    def parse_analysis(self, qc_metrics_raw: list[MetricsBase], **kwargs) -> NextflowAnalysis:
+    def parse_analysis(
+        self, qc_metrics_raw: list[MetricsBase], workflow: str, **kwargs
+    ) -> NextflowAnalysis:
         """Parse Nextflow output analysis files and return an analysis model."""
         sample_metrics: dict[str, dict] = {}
         for metric in qc_metrics_raw:
             try:
                 sample_metrics[metric.id].update({metric.name.lower(): metric.value})
             except KeyError:
                 sample_metrics[metric.id] = {metric.name.lower(): metric.value}
+            sample_metrics[metric.id]["type"] = workflow
         return NextflowAnalysis(sample_metrics=sample_metrics)
 
     def get_latest_metadata(self, case_id: str) -> NextflowAnalysis:
         """Return analysis output of a Nextflow case."""
         qc_metrics: list[MetricsBase] = self.get_multiqc_json_metrics(case_id)
-        return self.parse_analysis(qc_metrics_raw=qc_metrics)
+        case: Case = self.status_db.get_case_by_internal_id(case_id)
+        workflow: str = case.data_analysis
+        return self.parse_analysis(qc_metrics_raw=qc_metrics, workflow=workflow)
 
     def clean_past_run_dirs(self, before_date: str, skip_confirmation: bool = False) -> None:
         """Clean past run directories"""

cg/models/analysis.py

@@ -1,10 +1,17 @@
-from pydantic import BaseModel, ConfigDict
+from typing import Annotated
+
+from pydantic import BaseModel, ConfigDict, Field
 
 from cg.models.raredisease.raredisease import RarediseaseQCMetrics
 from cg.models.rnafusion.rnafusion import RnafusionQCMetrics
 from cg.models.taxprofiler.taxprofiler import TaxprofilerQCMetrics
 from cg.models.tomte.tomte import TomteQCMetrics
 
+DiscriminatedMetric = Annotated[
+    RarediseaseQCMetrics | RnafusionQCMetrics | TaxprofilerQCMetrics | TomteQCMetrics,
+    Field(discriminator="type"),
+]
+
 
 class AnalysisModel(BaseModel):
     """Metadata analysis model"""
@@ -15,6 +22,4 @@ class AnalysisModel(BaseModel):
 class NextflowAnalysis(AnalysisModel):
     """Nextflow's analysis results model."""
 
-    sample_metrics: dict[
-        str, RarediseaseQCMetrics | RnafusionQCMetrics | TaxprofilerQCMetrics | TomteQCMetrics
-    ]
+    sample_metrics: dict[str, DiscriminatedMetric]

cg/models/raredisease/raredisease.py

@@ -1,6 +1,9 @@
 from enum import StrEnum
+from typing import Literal
 
-from cg.constants.constants import SexOptions
+from pydantic import Field
+
+from cg.constants.constants import SexOptions, Workflow
 from cg.models.nf_analysis import NextflowSampleSheetEntry, WorkflowParameters
 from cg.models.qc_metrics import QCMetrics
 
@@ -12,6 +15,7 @@ class RarediseaseQCMetrics(QCMetrics):
     percent_duplicates: float
     predicted_sex_sex_check: SexOptions
     total_reads: int
+    type: Literal[Workflow.RAREDISEASE]
 
 
 class RarediseaseSampleSheetEntry(NextflowSampleSheetEntry):

cg/models/rnafusion/rnafusion.py

@@ -1,4 +1,6 @@
-from cg.constants.constants import Strandedness
+from typing import Literal
+
+from cg.constants.constants import Strandedness, Workflow
 from cg.models.nf_analysis import NextflowSampleSheetEntry, WorkflowParameters
 from cg.models.qc_metrics import QCMetrics
 
@@ -19,6 +21,7 @@ class RnafusionQCMetrics(QCMetrics):
     pct_duplication: float
     read_pairs_examined: float
     uniquely_mapped_percent: float
+    type: Literal[Workflow.RNAFUSION]
 
 
 class RnafusionParameters(WorkflowParameters):

cg/models/taxprofiler/taxprofiler.py

@@ -1,12 +1,15 @@
 from pathlib import Path
+from typing import Literal
 
-from pydantic import BaseModel, Field
+from pydantic import Field
 
+from cg.constants import Workflow
 from cg.constants.sequencing import SequencingPlatform
 from cg.models.nf_analysis import NextflowSampleSheetEntry, WorkflowParameters
+from cg.models.qc_metrics import QCMetrics
 
 
-class TaxprofilerQCMetrics(BaseModel):
+class TaxprofilerQCMetrics(QCMetrics):
     """Taxprofiler QC metrics."""
 
     after_filtering_gc_content: float
@@ -17,6 +20,7 @@ class TaxprofilerQCMetrics(BaseModel):
     pct_duplication: float
     raw_total_sequences: float
     reads_mapped: float
+    type: Literal[Workflow.TAXPROFILER]
 
 
 class TaxprofilerParameters(WorkflowParameters):

cg/models/tomte/tomte.py

@@ -1,9 +1,10 @@
 from enum import StrEnum
 from pathlib import Path
+from typing import Literal
 
 from pydantic import field_validator
 
-from cg.constants.constants import GenomeVersion, Strandedness
+from cg.constants.constants import GenomeVersion, Strandedness, Workflow
 from cg.constants.sample_sources import SourceType
 from cg.models.nf_analysis import NextflowSampleSheetEntry, WorkflowParameters
 from cg.models.qc_metrics import QCMetrics
@@ -87,3 +88,4 @@ class TomteQCMetrics(QCMetrics):
     pct_ribosomal_bases: float
     pct_surviving: float
     uniquely_mapped_percent: float
+    type: Literal[Workflow.TOMTE]

tests/meta/workflow/conftest.py

@@ -281,6 +281,7 @@ def rnafusion_metrics() -> dict[str, float]:
         "pct_duplication": 14.8643,
         "read_pairs_examined": 72391566.0,
         "uniquely_mapped_percent": 91.02,
+        "type": Workflow.RNAFUSION,
     }
 
 

tests/meta/workflow/test_rnafusion.py

@@ -1,5 +1,6 @@
 """Module for Rnafusion analysis API tests."""
 
+from cg.constants import Workflow
 from cg.meta.workflow.rnafusion import RnafusionAnalysisAPI
 from cg.models.analysis import NextflowAnalysis
 from cg.models.cg_config import CGConfig
@@ -21,7 +22,9 @@ def test_parse_analysis(
     qc_metrics: list[MetricsBase] = analysis_api.get_multiqc_json_metrics(case_id=rnafusion_case_id)
 
     # WHEN extracting the analysis model
-    analysis_model: NextflowAnalysis = analysis_api.parse_analysis(qc_metrics_raw=qc_metrics)
+    analysis_model: NextflowAnalysis = analysis_api.parse_analysis(
+        qc_metrics_raw=qc_metrics, workflow=Workflow.RNAFUSION
+    )
 
     # THEN the analysis model and its content should have been correctly extracted
     assert analysis_model.sample_metrics[sample_id].model_dump() == rnafusion_metrics