try sharing these params?

Author: bpblanken

v03_pipeline/lib/tasks/base/base_loading_params.py

@@ -0,0 +1,32 @@
+import luigi
+
+from v03_pipeline.lib.model import SampleType
+
+
+class BaseLoadingParams(luigi.Task):
+    # NB:
+    # These params are "inherited" with the special
+    # luigi.util.inherits function, copying params
+    # but nothing else.
+    sample_type = luigi.EnumParameter(enum=SampleType)
+    callset_path = luigi.Parameter()
+    imputed_sex_path = luigi.Parameter(
+        default=None,
+        description='Optional path to a tsv of imputed sex values from the DRAGEN GVS pipeline.',
+    )
+    filters_path = luigi.Parameter(
+        default=None,
+        description='Optional path to part two outputs from callset (VCF shards containing filter information)',
+    )
+    validate = luigi.BoolParameter(
+        default=True,
+        parsing=luigi.BoolParameter.EXPLICIT_PARSING,
+    )
+    force = luigi.BoolParameter(
+        default=False,
+        parsing=luigi.BoolParameter.EXPLICIT_PARSING,
+    )
+    check_sex_and_relatedness = luigi.BoolParameter(
+        default=False,
+        parsing=luigi.BoolParameter.EXPLICIT_PARSING,
+    )

v03_pipeline/lib/tasks/write_imported_callset.py

@@ -15,39 +15,25 @@
     validate_sample_type,
 )
 from v03_pipeline.lib.misc.vets import annotate_vets
-from v03_pipeline.lib.model import CachedReferenceDatasetQuery, SampleType
+from v03_pipeline.lib.model import CachedReferenceDatasetQuery
 from v03_pipeline.lib.model.environment import Env
 from v03_pipeline.lib.paths import (
     cached_reference_dataset_query_path,
     imported_callset_path,
     sex_check_table_path,
 )
+from v03_pipeline.lib.tasks.base.base_loading_params import BaseLoadingParams
 from v03_pipeline.lib.tasks.base.base_write import BaseWriteTask
 from v03_pipeline.lib.tasks.files import CallsetTask, GCSorLocalTarget, HailTableTask
 from v03_pipeline.lib.tasks.reference_data.updated_cached_reference_dataset_query import (
     UpdatedCachedReferenceDatasetQuery,
 )
 from v03_pipeline.lib.tasks.write_sex_check_table import WriteSexCheckTableTask
 
+luigi.util.inherits(BaseLoadingParams)
 
-class WriteImportedCallsetTask(BaseWriteTask):
-    sample_type = luigi.EnumParameter(enum=SampleType)
-    callset_path = luigi.Parameter()
-    imputed_sex_path = luigi.Parameter(default=None)
-    filters_path = luigi.OptionalParameter(default=None)
-    validate = luigi.BoolParameter(
-        default=True,
-        parsing=luigi.BoolParameter.EXPLICIT_PARSING,
-    )
-    force = luigi.BoolParameter(
-        default=False,
-        parsing=luigi.BoolParameter.EXPLICIT_PARSING,
-    )
-    check_sex_and_relatedness = luigi.BoolParameter(
-        default=False,
-        parsing=luigi.BoolParameter.EXPLICIT_PARSING,
-    )
 
+class WriteImportedCallsetTask(BaseWriteTask):
     def complete(self) -> luigi.Target:
         if not self.force and super().complete():
             mt = hl.read_matrix_table(self.output().path)
@@ -76,7 +62,10 @@ def requires(self) -> list[luigi.Task]:
             requirements = [
                 *requirements,
                 (
-                    self.clone(UpdatedCachedReferenceDatasetQuery, crdq=CachedReferenceDatasetQuery.GNOMAD_CODING_AND_NONCODING_VARIANTS),
+                    self.clone(
+                        UpdatedCachedReferenceDatasetQuery,
+                        crdq=CachedReferenceDatasetQuery.GNOMAD_CODING_AND_NONCODING_VARIANTS,
+                    )
                     if Env.REFERENCE_DATA_AUTO_UPDATE
                     else HailTableTask(
                         cached_reference_dataset_query_path(

v03_pipeline/lib/tasks/write_relatedness_check_table.py

@@ -33,7 +33,6 @@ def requires(self) -> luigi.Task:
             WriteImportedCallsetTask(
                 self.reference_genome,
                 self.dataset_type,
-                self.sample_type,
                 self.callset_path,
             ),
         ]

v03_pipeline/lib/tasks/write_sex_check_table.py

@@ -21,9 +21,7 @@ def output(self) -> luigi.Target:
         )
 
     def requires(self) -> luigi.Task:
-        return [
-            RawFileTask(self.imputed_sex_path),
-        ]
+        return RawFileTask(self.imputed_sex_path)
 
     def create_table(self) -> hl.Table:
-        return import_imputed_sex(self.input()[0].path)
+        return import_imputed_sex(self.input().path)