Merge branch 'benb/remove_reference_data_env_var' of github.com:broadinstitute/seqr-loading-pipelines into dev

Author: bpblanken

v03_pipeline/lib/tasks/base/base_delete_table.py

@@ -1,13 +1,21 @@
 import hailtop.fs as hfs
+import luigi
+import luigi.util
 
 from v03_pipeline.lib.logger import get_logger
-from v03_pipeline.lib.tasks.base.base_hail_table import BaseHailTableTask
+from v03_pipeline.lib.model import SampleType
+from v03_pipeline.lib.tasks.base.base_loading_pipeline_params import (
+    BaseLoadingPipelineParams,
+)
 from v03_pipeline.lib.tasks.files import GCSorLocalFolderTarget, GCSorLocalTarget
 
 logger = get_logger(__name__)
 
 
-class BaseDeleteTableTask(BaseHailTableTask):
+@luigi.util.inherits(BaseLoadingPipelineParams)
+class BaseDeleteTableTask(luigi.Task):
+    sample_type = luigi.EnumParameter(enum=SampleType)
+
     def complete(self) -> bool:
         logger.info(f'DeleteTableTask: checking if {self.output().path} exists')
         return (

v03_pipeline/lib/tasks/base/base_hail_table.py

@@ -2,16 +2,17 @@
 import luigi
 
 from v03_pipeline.lib.logger import get_logger
-from v03_pipeline.lib.model import DatasetType, Env, ReferenceGenome
+from v03_pipeline.lib.model import Env
+from v03_pipeline.lib.tasks.base.base_loading_pipeline_params import (
+    BaseLoadingPipelineParams,
+)
 from v03_pipeline.lib.tasks.files import GCSorLocalFolderTarget
 
 logger = get_logger(__name__)
 
 
+@luigi.util.inherits(BaseLoadingPipelineParams)
 class BaseHailTableTask(luigi.Task):
-    reference_genome = luigi.EnumParameter(enum=ReferenceGenome)
-    dataset_type = luigi.EnumParameter(enum=DatasetType)
-
     def output(self) -> luigi.Target:
         raise NotImplementedError
 

v03_pipeline/lib/tasks/base/base_loading_pipeline_params.py

@@ -0,0 +1,12 @@
+import luigi
+
+from v03_pipeline.lib.model import DatasetType, ReferenceGenome
+
+
+class BaseLoadingPipelineParams(luigi.Task):
+    # NB:
+    # These params are "inherited" with the special
+    # luigi.util.inherits function, copying params
+    # but nothing else.
+    reference_genome = luigi.EnumParameter(enum=ReferenceGenome)
+    dataset_type = luigi.EnumParameter(enum=DatasetType)

v03_pipeline/lib/tasks/base/base_loading_run_params.py

@@ -1,13 +1,21 @@
 import luigi
 
 from v03_pipeline.lib.model import SampleType
+from v03_pipeline.lib.tasks.base.base_loading_pipeline_params import (
+    BaseLoadingPipelineParams,
+)
 
 
+@luigi.util.inherits(BaseLoadingPipelineParams)
 class BaseLoadingRunParams(luigi.Task):
-    # NB:
-    # These params are "inherited" with the special
-    # luigi.util.inherits function, copying params
-    # but nothing else.
+    # The difference between the "Loading Run" params
+    # and the "Loading Pipeline" params:
+    # - These params are used during standard "runs"
+    # of the pipeline that add a callset to the backend
+    # data store.
+    # - The "Loading Pipeline" params are shared with
+    # tasks that may remove data from or change the
+    # structure of the persisted Hail Tables.
     sample_type = luigi.EnumParameter(enum=SampleType)
     callset_path = luigi.Parameter()
     ignore_missing_samples_when_remapping = luigi.BoolParameter(

v03_pipeline/lib/tasks/delete_family_table.py

@@ -1,13 +1,11 @@
 import luigi
 
-from v03_pipeline.lib.model import SampleType
 from v03_pipeline.lib.paths import family_table_path
 from v03_pipeline.lib.tasks.base.base_delete_table import BaseDeleteTableTask
 from v03_pipeline.lib.tasks.files import GCSorLocalTarget
 
 
 class DeleteFamilyTableTask(BaseDeleteTableTask):
-    sample_type = luigi.EnumParameter(enum=SampleType)
     family_guid = luigi.Parameter()
 
     def output(self) -> luigi.Target:

v03_pipeline/lib/tasks/delete_family_tables.py

@@ -1,11 +1,14 @@
 import luigi
 
 from v03_pipeline.lib.model import SampleType
-from v03_pipeline.lib.tasks.base.base_hail_table import BaseHailTableTask
+from v03_pipeline.lib.tasks.base.base_loading_pipeline_params import (
+    BaseLoadingPipelineParams,
+)
 from v03_pipeline.lib.tasks.delete_family_table import DeleteFamilyTableTask
 
 
-class DeleteFamilyTablesTask(BaseHailTableTask):
+@luigi.util.inherits(BaseLoadingPipelineParams)
+class DeleteFamilyTablesTask(luigi.Task):
     family_guids = luigi.ListParameter()
 
     def __init__(self, *args, **kwargs):

v03_pipeline/lib/tasks/delete_project_family_tables.py

@@ -1,15 +1,19 @@
 import hail as hl
 import hailtop.fs as hfs
 import luigi
+import luigi.util
 
 from v03_pipeline.lib.model import SampleType
 from v03_pipeline.lib.paths import project_table_path
-from v03_pipeline.lib.tasks.base.base_hail_table import BaseHailTableTask
+from v03_pipeline.lib.tasks.base.base_loading_pipeline_params import (
+    BaseLoadingPipelineParams,
+)
 from v03_pipeline.lib.tasks.delete_family_table import DeleteFamilyTableTask
 from v03_pipeline.lib.tasks.files import HailTableTask
 
 
-class DeleteProjectFamilyTablesTask(BaseHailTableTask):
+@luigi.util.inherits(BaseLoadingPipelineParams)
+class DeleteProjectFamilyTablesTask(luigi.Task):
     project_guid = luigi.Parameter()
 
     def __init__(self, *args, **kwargs):

v03_pipeline/lib/tasks/delete_project_table.py

@@ -1,6 +1,5 @@
 import luigi
 
-from v03_pipeline.lib.model import SampleType
 from v03_pipeline.lib.paths import project_table_path
 from v03_pipeline.lib.tasks.base.base_delete_table import BaseDeleteTableTask
 from v03_pipeline.lib.tasks.delete_project_family_tables import (
@@ -10,7 +9,6 @@
 
 
 class DeleteProjectTableTask(BaseDeleteTableTask):
-    sample_type = luigi.EnumParameter(enum=SampleType)
     project_guid = luigi.Parameter()
 
     def requires(self) -> luigi.Task:

v03_pipeline/lib/tasks/reference_data/update_cached_reference_dataset_queries.py

@@ -1,19 +1,19 @@
 import luigi
+import luigi.util
 
 from v03_pipeline.lib.model import (
     CachedReferenceDatasetQuery,
-    DatasetType,
-    ReferenceGenome,
+)
+from v03_pipeline.lib.tasks.base.base_loading_pipeline_params import (
+    BaseLoadingPipelineParams,
 )
 from v03_pipeline.lib.tasks.reference_data.updated_cached_reference_dataset_query import (
     UpdatedCachedReferenceDatasetQuery,
 )
 
 
+@luigi.util.inherits(BaseLoadingPipelineParams)
 class UpdateCachedReferenceDatasetQueries(luigi.Task):
-    reference_genome = luigi.EnumParameter(enum=ReferenceGenome)
-    dataset_type = luigi.EnumParameter(enum=DatasetType)
-
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.checked_for_tasks = False

v03_pipeline/lib/tasks/write_metadata_for_run.py

@@ -5,7 +5,6 @@
 import luigi.util
 
 from v03_pipeline.lib.paths import metadata_for_run_path
-from v03_pipeline.lib.tasks.base.base_hail_table import BaseHailTableTask
 from v03_pipeline.lib.tasks.base.base_loading_run_params import BaseLoadingRunParams
 from v03_pipeline.lib.tasks.files import GCSorLocalTarget
 from v03_pipeline.lib.tasks.write_remapped_and_subsetted_callset import (
@@ -14,7 +13,7 @@
 
 
 @luigi.util.inherits(BaseLoadingRunParams)
-class WriteMetadataForRunTask(BaseHailTableTask):
+class WriteMetadataForRunTask(luigi.Task):
     project_guids = luigi.ListParameter()
     project_remap_paths = luigi.ListParameter()
     project_pedigree_paths = luigi.ListParameter()