Sample sheet api from context

cg/apps/demultiplex/sample_sheet/api.py

@@ -18,7 +18,6 @@
 from cg.meta.demultiplex.housekeeper_storage_functions import (
     add_and_include_sample_sheet_path_to_housekeeper,
 )
-from cg.models.cg_config import CGConfig
 from cg.models.flow_cell.flow_cell import FlowCellDirectoryData
 
 LOG = logging.getLogger(__name__)
@@ -27,11 +26,10 @@
 class SampleSheetAPI:
     """Sample Sheet API class."""
 
-    def __init__(self, config: CGConfig) -> None:
-        self.config: CGConfig = config
-        self.flow_cell_runs_dir = Path(config.illumina_flow_cells_directory)
-        self.hk_api: HousekeeperAPI = config.housekeeper_api
-        self.lims_api: LimsAPI = config.lims_api
+    def __init__(self, flow_cell_dir: str, hk_api: HousekeeperAPI, lims_api: LimsAPI) -> None:
+        self.flow_cell_runs_dir = Path(flow_cell_dir)
+        self.hk_api: HousekeeperAPI = hk_api
+        self.lims_api: LimsAPI = lims_api
         self.dry_run: bool = False
         self.force: bool = False
         self.validator = SampleSheetValidator()

cg/cli/demultiplex/demux.py

@@ -37,7 +37,7 @@
 def demultiplex_all(context: CGConfig, flow_cells_directory: click.Path, dry_run: bool):
     """Demultiplex all flow cells that are ready under the flow cells directory."""
     LOG.info("Running cg demultiplex all ...")
-    sample_sheet_api = SampleSheetAPI(context)
+    sample_sheet_api: SampleSheetAPI = context.sample_sheet_api
     demultiplex_api: DemultiplexingAPI = context.demultiplex_api
     demultiplex_api.set_dry_run(dry_run=dry_run)
     if flow_cells_directory:
@@ -94,7 +94,7 @@ def demultiplex_flow_cell(
     """
 
     LOG.info(f"Running cg demultiplex flow cell, using {bcl_converter}")
-    sample_sheet_api = SampleSheetAPI(context)
+    sample_sheet_api: SampleSheetAPI = context.sample_sheet_api
     demultiplex_api: DemultiplexingAPI = context.demultiplex_api
     flow_cell_directory: Path = Path(context.demultiplex_api.flow_cells_dir, flow_cell_name)
     demultiplex_api.set_dry_run(dry_run=dry_run)

cg/cli/demultiplex/sample_sheet.py

@@ -24,7 +24,7 @@ def sample_sheet_commands():
 def validate_sample_sheet(context: CGConfig, sheet: click.Path):
     """Validate a sample sheet."""
     LOG.info(f"Validating {sheet} sample sheet")
-    sample_sheet_api = SampleSheetAPI(config=context)
+    sample_sheet_api: SampleSheetAPI = context.sample_sheet_api
     try:
         sample_sheet_api.validate(Path(sheet))
     except (SampleSheetError, ValidationError) as error:
@@ -49,7 +49,7 @@ def create_sheet(
     Search the flow cell in the directory specified in config.
     """
     LOG.info(f"Creating sample sheet for flow cell {flow_cell_name}")
-    sample_sheet_api = SampleSheetAPI(context)
+    sample_sheet_api: SampleSheetAPI = context.sample_sheet_api
     sample_sheet_api.set_dry_run(dry_run)
     sample_sheet_api.set_force(force)
     sample_sheet_api.get_or_create_sample_sheet(
@@ -66,7 +66,7 @@ def create_all_sheets(context: CGConfig, dry_run: bool):
     Search flow cell directories for run parameters and create a sample sheets based on the
     information.
     """
-    sample_sheet_api = SampleSheetAPI(context)
+    sample_sheet_api: SampleSheetAPI = context.sample_sheet_api
     sample_sheet_api.set_dry_run(dry_run)
     sample_sheet_api.set_force(force=False)
     sample_sheet_api.get_or_create_all_sample_sheets()

cg/models/cg_config.py

@@ -6,6 +6,7 @@
 from cg.apps.coverage import ChanjoAPI
 from cg.apps.crunchy import CrunchyAPI
 from cg.apps.demultiplex.demultiplex_api import DemultiplexingAPI
+from cg.apps.demultiplex.sample_sheet.api import SampleSheetAPI
 from cg.apps.gens import GensAPI
 from cg.apps.gt import GenotypeAPI
 from cg.apps.hermes.hermes_api import HermesApi
@@ -307,6 +308,7 @@ class CGConfig(BaseModel):
     pigz: CommonAppConfig | None = None
     pdc: CommonAppConfig | None = None
     pdc_api_: PdcAPI | None
+    sample_sheet_api_: SampleSheetAPI | None = None
     scout: CommonAppConfig = None
     scout_api_: ScoutAPI = None
     tar: CommonAppConfig | None = None
@@ -463,6 +465,19 @@ def pdc_api(self) -> PdcAPI:
             self.pdc_api_ = api
         return api
 
+    @property
+    def sample_sheet_api(self) -> SampleSheetAPI:
+        sample_sheet_api = self.__dict__.get("sample_sheet_api_")
+        if sample_sheet_api is None:
+            LOG.debug("Instantiating sample sheet API")
+            sample_sheet_api = SampleSheetAPI(
+                flow_cell_dir=self.illumina_flow_cells_directory,
+                hk_api=self.housekeeper_api,
+                lims_api=self.lims_api,
+            )
+            self.sample_sheet_api_ = sample_sheet_api
+        return sample_sheet_api
+
     @property
     def slurm_service(self) -> SlurmService:
         return SlurmCLIService()

tests/cli/demultiplex/test_create_sample_sheet.py

@@ -10,6 +10,8 @@
     FlowCellSampleBcl2Fastq,
     FlowCellSampleBCLConvert,
 )
+from cg.apps.housekeeper.hk import HousekeeperAPI
+from cg.apps.lims import LimsAPI
 from cg.cli.demultiplex.sample_sheet import create_sheet
 from cg.constants.demultiplexing import BclConverter
 from cg.constants.process import EXIT_SUCCESS
@@ -23,7 +25,7 @@
 def test_create_sample_sheet_no_run_parameters_fails(
     cli_runner: testing.CliRunner,
     tmp_flow_cell_without_run_parameters_path: Path,
-    sample_sheet_context: CGConfig,
+    sample_sheet_context_broken_flow_cells: CGConfig,
     hiseq_2500_custom_index_bcl_convert_lims_samples: list[FlowCellSampleBCLConvert],
     caplog,
     mocker,
@@ -34,20 +36,21 @@ def test_create_sample_sheet_no_run_parameters_fails(
         flow_cell_path=tmp_flow_cell_without_run_parameters_path
     )
 
+    # GIVEN that the context's flow cell directory holds the given flow cell
+    assert (
+        sample_sheet_context_broken_flow_cells.illumina_demultiplexed_runs_directory
+        == flow_cell.path.parent.as_posix()
+    )
+
     # GIVEN flow cell samples
     mocker.patch(
         FLOW_CELL_FUNCTION_NAME,
         return_value=hiseq_2500_custom_index_bcl_convert_lims_samples,
     )
 
-    # GIVEN that the context's flow cell directory holds the given flow cell
-    sample_sheet_context.illumina_flow_cells_directory = (
-        tmp_flow_cell_without_run_parameters_path.parent.as_posix()
-    )
-
     # WHEN running the create sample sheet command
     result: testing.Result = cli_runner.invoke(
-        create_sheet, [flow_cell.full_name], obj=sample_sheet_context
+        create_sheet, [flow_cell.full_name], obj=sample_sheet_context_broken_flow_cells
     )
 
     # THEN the process exits with a non-zero exit code
@@ -85,7 +88,8 @@ def test_create_bcl2fastq_sample_sheet(
         FLOW_CELL_FUNCTION_NAME,
         return_value=novaseq_6000_pre_1_5_kits_bcl2fastq_lims_samples,
     )
-    # GIVEN a lims api that returns some samples
+    # GIVEN a sample sheet API and a lims API that returns some samples
+    sample_sheet_api: SampleSheetAPI = sample_sheet_context.sample_sheet_api
 
     # WHEN creating a sample sheet
     result = cli_runner.invoke(
@@ -105,7 +109,7 @@ def test_create_bcl2fastq_sample_sheet(
     assert flow_cell.sample_sheet_exists()
 
     # THEN the sample sheet passes validation
-    SampleSheetAPI(sample_sheet_context).validate(flow_cell.sample_sheet_path)
+    sample_sheet_api.validate(flow_cell.sample_sheet_path)
 
     # THEN the sample sheet is in Housekeeper
     assert sample_sheet_context.housekeeper_api.get_sample_sheets_from_latest_version(flow_cell.id)
@@ -146,8 +150,11 @@ def test_create_v2_sample_sheet(
     request: FixtureRequest,
 ):
     """Test that creating a v2 sample sheet works."""
-    flow_cell_directory: Path = request.getfixturevalue(scenario.flow_cell_directory)
+    # GIVEN a sample sheet context with a sample sheet api
+    sample_sheet_api: SampleSheetAPI = sample_sheet_context.sample_sheet_api
+
     # GIVEN a flow cell directory with some run parameters
+    flow_cell_directory: Path = request.getfixturevalue(scenario.flow_cell_directory)
     flow_cell: FlowCellDirectoryData = FlowCellDirectoryData(flow_cell_directory)
     assert flow_cell.run_parameters_path.exists()
 
@@ -180,7 +187,7 @@ def test_create_v2_sample_sheet(
     assert flow_cell.sample_sheet_exists()
 
     # THEN the sample sheet passes validation
-    SampleSheetAPI(sample_sheet_context).validate(flow_cell.sample_sheet_path)
+    sample_sheet_api.validate(flow_cell.sample_sheet_path)
 
     # THEN the sample sheet is in Housekeeper
     assert sample_sheet_context.housekeeper_api.get_sample_sheets_from_latest_version(flow_cell.id)
@@ -207,7 +214,7 @@ def test_incorrect_bcl2fastq_samplesheet_is_regenerated(
     )
 
     # GIVEN a sample sheet API and an invalid sample sheet
-    sample_sheet_api: SampleSheetAPI = SampleSheetAPI(sample_sheet_context)
+    sample_sheet_api: SampleSheetAPI = sample_sheet_context.sample_sheet_api
     with pytest.raises(ValidationError):
         sample_sheet_api.validate(flow_cell.sample_sheet_path)
 

tests/conftest.py

@@ -17,6 +17,7 @@
 
 from cg.apps.crunchy import CrunchyAPI
 from cg.apps.demultiplex.demultiplex_api import DemultiplexingAPI
+from cg.apps.demultiplex.sample_sheet.api import SampleSheetAPI
 from cg.apps.downsample.downsample import DownsampleAPI
 from cg.apps.gens import GensAPI
 from cg.apps.gt import GenotypeAPI
@@ -436,11 +437,38 @@ def gens_config() -> dict[str, dict[str, str]]:
 
 @pytest.fixture(name="sample_sheet_context")
 def sample_sheet_context(
-    cg_context: CGConfig, lims_api: LimsAPI, populated_housekeeper_api: HousekeeperAPI
+    cg_context: CGConfig,
+    lims_api: LimsAPI,
+    populated_housekeeper_api: HousekeeperAPI,
+    tmp_illumina_flow_cells_directory: Path,
 ) -> CGConfig:
     """Return cg context with added Lims and Housekeeper API."""
     cg_context.lims_api_ = lims_api
     cg_context.housekeeper_api_ = populated_housekeeper_api
+    cg_context.sample_sheet_api_ = SampleSheetAPI(
+        flow_cell_dir=tmp_illumina_flow_cells_directory.as_posix(),
+        hk_api=cg_context.housekeeper_api,
+        lims_api=cg_context.lims_api,
+    )
+    return cg_context
+
+
+@pytest.fixture
+def sample_sheet_context_broken_flow_cells(
+    cg_context: CGConfig,
+    lims_api: LimsAPI,
+    populated_housekeeper_api: HousekeeperAPI,
+    tmp_broken_flow_cells_directory: Path,
+) -> CGConfig:
+    """Return cg context with broken flow cells."""
+    cg_context.illumina_demultiplexed_runs_directory = tmp_broken_flow_cells_directory.as_posix()
+    cg_context.lims_api_ = lims_api
+    cg_context.housekeeper_api_ = populated_housekeeper_api
+    cg_context.sample_sheet_api_ = SampleSheetAPI(
+        flow_cell_dir=tmp_broken_flow_cells_directory.as_posix(),
+        hk_api=cg_context.housekeeper_api,
+        lims_api=cg_context.lims_api,
+    )
     return cg_context