refactor to ensure we do all the tasks before writing a success file

Author: bpblanken

v03_pipeline/bin/pipeline_worker.py

@@ -13,9 +13,6 @@
     project_pedigree_path,
     project_remap_path,
 )
-from v03_pipeline.lib.tasks import (
-    UpdateVariantAnnotationsTableWithNewSamplesTask,
-)
 from v03_pipeline.lib.tasks.trigger_hail_backend_reload import TriggerHailBackendReload
 from v03_pipeline.lib.tasks.write_success_file import WriteSuccessFileTask
 
@@ -57,16 +54,14 @@ def main():
                 'run_id': run_id,
                 **{k: v for k, v in lpr.model_dump().items() if k != 'projects_to_run'},
             }
-            tasks = [
-                UpdateVariantAnnotationsTableWithNewSamplesTask(
-                    **loading_run_task_params,
-                ),
-                WriteSuccessFileTask(**loading_run_task_params),
-            ]
             if Env.SHOULD_TRIGGER_HAIL_BACKEND_RELOAD:
-                tasks.append(
+                tasks = [
                     TriggerHailBackendReload(**loading_run_task_params),
-                )
+                ]
+            else:
+                tasks = [
+                    WriteSuccessFileTask(**loading_run_task_params),
+                ]
             luigi.build(tasks)
         except Exception:
             logger.exception('Unhandled Exception')

v03_pipeline/lib/tasks/trigger_hail_backend_reload.py

@@ -4,10 +4,10 @@
 
 from v03_pipeline.lib.logger import get_logger
 from v03_pipeline.lib.model import Env
-from v03_pipeline.lib.tasks import UpdateVariantAnnotationsTableWithNewSamplesTask
 from v03_pipeline.lib.tasks.base.base_project_info_params import (
     BaseLoadingRunWithProjectInfoParams,
 )
+from v03_pipeline.lib.tasks.write_success_file import WriteSuccessFileTask
 
 logger = get_logger(__name__)
 
@@ -19,7 +19,7 @@ def __init__(self, *args, **kwargs):
         self.done = False
 
     def requires(self):
-        return self.clone(UpdateVariantAnnotationsTableWithNewSamplesTask)
+        return self.clone(WriteSuccessFileTask)
 
     def run(self):
         url = f'http://{Env.HAIL_BACKEND_SERVICE_HOSTNAME}:{Env.HAIL_BACKEND_SERVICE_PORT}/reload_globals'

v03_pipeline/lib/tasks/write_success_file.py

@@ -7,6 +7,9 @@
     BaseLoadingRunWithProjectInfoParams,
 )
 from v03_pipeline.lib.tasks.files import GCSorLocalTarget
+from v03_pipeline.lib.tasks.update_variant_annotations_table_with_new_samples import (
+    UpdateVariantAnnotationsTableWithNewSamplesTask,
+)
 
 
 @luigi.util.inherits(BaseLoadingRunWithProjectInfoParams)
@@ -21,14 +24,20 @@ def output(self) -> luigi.Target:
         )
 
     def requires(self):
+        requirements = [
+            self.clone(UpdateVariantAnnotationsTableWithNewSamplesTask),
+        ]
         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))
+            *requirements,
+            *[
+                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):