add task to write success file for run (#920)

* add task to write success file for run

* remove WriteProjectFamilyTablesTask from pipeline_worker

* no hfs

* add to init

Author: jklugherz

v03_pipeline/bin/pipeline_worker.py

@@ -15,9 +15,9 @@
 )
 from v03_pipeline.lib.tasks import (
     UpdateVariantAnnotationsTableWithNewSamplesTask,
-    WriteProjectFamilyTablesTask,
 )
 from v03_pipeline.lib.tasks.trigger_hail_backend_reload import TriggerHailBackendReload
+from v03_pipeline.lib.tasks.write_success_file import WriteSuccessFileTask
 
 logger = get_logger(__name__)
 
@@ -47,39 +47,25 @@ def main():
                 )
                 for project_guid in lpr.projects_to_run
             ]
-            task_kwargs = {
-                k: v for k, v in lpr.model_dump().items() if k != 'projects_to_run'
-            }
             run_id = datetime.datetime.now(datetime.timezone.utc).strftime(
                 '%Y%m%d-%H%M%S',
             )
+            loading_run_task_params = {
+                'project_guids': lpr.projects_to_run,
+                'project_remap_paths': project_remap_paths,
+                'project_pedigree_paths': project_pedigree_paths,
+                'run_id': run_id,
+                **{k: v for k, v in lpr.model_dump().items() if k != 'projects_to_run'},
+            }
             tasks = [
                 UpdateVariantAnnotationsTableWithNewSamplesTask(
-                    project_guids=lpr.projects_to_run,
-                    project_remap_paths=project_remap_paths,
-                    project_pedigree_paths=project_pedigree_paths,
-                    run_id=run_id,
-                    **task_kwargs,
+                    **loading_run_task_params,
                 ),
-                *[
-                    WriteProjectFamilyTablesTask(
-                        project_guid=lpr.projects_to_run[i],
-                        project_remap_path=project_remap_paths[i],
-                        project_pedigree_path=project_pedigree_paths[i],
-                        **task_kwargs,
-                    )
-                    for i in range(len(lpr.projects_to_run))
-                ],
+                WriteSuccessFileTask(**loading_run_task_params),
             ]
             if Env.SHOULD_TRIGGER_HAIL_BACKEND_RELOAD:
                 tasks.append(
-                    TriggerHailBackendReload(
-                        project_guids=lpr.projects_to_run,
-                        project_remap_paths=project_remap_paths,
-                        project_pedigree_paths=project_pedigree_paths,
-                        run_id=run_id,
-                        **task_kwargs,
-                    ),
+                    TriggerHailBackendReload(**loading_run_task_params),
                 )
             luigi.build(tasks)
         except Exception:

v03_pipeline/lib/paths.py

@@ -366,3 +366,18 @@ def loading_pipeline_queue_path() -> str:
         'loading_pipeline_queue',
         'request.json',
     )
+
+
+def pipeline_run_success_file_path(
+    reference_genome: ReferenceGenome,
+    dataset_type: DatasetType,
+    run_id: str,
+) -> str:
+    return os.path.join(
+        runs_path(
+            reference_genome,
+            dataset_type,
+        ),
+        run_id,
+        '_SUCCESS',
+    )

v03_pipeline/lib/tasks/__init__.py

@@ -34,6 +34,7 @@
 from v03_pipeline.lib.tasks.write_project_family_tables import (
     WriteProjectFamilyTablesTask,
 )
+from v03_pipeline.lib.tasks.write_success_file import WriteSuccessFileTask
 
 __all__ = [
     'DeleteFamilyTableTask',
@@ -52,4 +53,5 @@
     'UpdateVariantAnnotationsTableWithDeletedFamiliesTask',
     'WriteMetadataForRunTask',
     'WriteProjectFamilyTablesTask',
+    'WriteSuccessFileTask',
 ]

v03_pipeline/lib/tasks/write_success_file.py

@@ -0,0 +1,36 @@
+import luigi
+import luigi.util
+
+from v03_pipeline.lib.paths import pipeline_run_success_file_path
+from v03_pipeline.lib.tasks import WriteProjectFamilyTablesTask
+from v03_pipeline.lib.tasks.base.base_project_info_params import (
+    BaseLoadingRunWithProjectInfoParams,
+)
+from v03_pipeline.lib.tasks.files import GCSorLocalTarget
+
+
+@luigi.util.inherits(BaseLoadingRunWithProjectInfoParams)
+class WriteSuccessFileTask(luigi.Task):
+    def output(self) -> luigi.Target:
+        return GCSorLocalTarget(
+            pipeline_run_success_file_path(
+                self.reference_genome,
+                self.dataset_type,
+                self.run_id,
+            ),
+        )
+
+    def requires(self):
+        return [
+            self.clone(
+                WriteProjectFamilyTablesTask,
+                project_guid=self.project_guids[i],
+                project_remap_path=self.project_remap_paths[i],
+                project_pedigree_path=self.project_pedigree_paths[i],
+            )
+            for i in range(len(self.project_guids))
+        ]
+
+    def run(self):
+        with self.output().open('w') as f:
+            f.write('')

v03_pipeline/lib/tasks/write_success_file_test.py

@@ -0,0 +1,33 @@
+from unittest import mock
+
+import luigi.worker
+
+from v03_pipeline.lib.model import DatasetType, ReferenceGenome, SampleType
+from v03_pipeline.lib.tasks.write_success_file import WriteSuccessFileTask
+from v03_pipeline.lib.test.mock_complete_task import MockCompleteTask
+from v03_pipeline.lib.test.mocked_dataroot_testcase import MockedDatarootTestCase
+
+
+class WriteSuccessFileTaskTest(MockedDatarootTestCase):
+    @mock.patch(
+        'v03_pipeline.lib.tasks.write_success_file.WriteProjectFamilyTablesTask',
+    )
+    def test_write_success_file_task(self, mock_write_project_fam_tables) -> None:
+        mock_write_project_fam_tables.return_value = MockCompleteTask()
+
+        worker = luigi.worker.Worker()
+        write_success_file = WriteSuccessFileTask(
+            reference_genome=ReferenceGenome.GRCh38,
+            dataset_type=DatasetType.SNV_INDEL,
+            sample_type=SampleType.WGS,
+            callset_path='test_callset',
+            project_guids=['R0113_test_project'],
+            project_remap_paths=['test_remap'],
+            project_pedigree_paths=['test_pedigree'],
+            run_id='manual__2024-04-03',
+        )
+        worker.add(write_success_file)
+        worker.run()
+        self.assertTrue(write_success_file.complete())
+        with open(write_success_file.output().path) as f:
+            self.assertEqual(f.read(), '')