chore: cleanup

Author: Szymon Szyszkowski

src/ot_orchestration/dags/gwas_catalog_harmonisation.py

@@ -9,9 +9,9 @@
 from airflow.models.baseoperator import chain
 from airflow.models.dag import DAG
 
-from ot_orchestration.operators.batch.harmonisation import (
+from ot_orchestration.operators.batch.generic import (
     BatchIndexOperator,
-    GeneticsBatchJobOperator,
+    BatchJobOperator,
 )
 from ot_orchestration.utils import (
     find_node_in_config,
@@ -50,7 +50,7 @@ def end():
         batch_index_specs=node_config["google_batch_index_specs"],
     )
     node_config = find_node_in_config(config["nodes"], "gwas_catalog_harmonisation")
-    harmonisation_batch_job = GeneticsBatchJobOperator.partial(
+    harmonisation_batch_job = BatchJobOperator.partial(
         task_id=node_config["id"],
         job_name="harmonisation",
         google_batch=node_config["google_batch"],

src/ot_orchestration/operators/batch/harmonisation.py renamed to src/ot_orchestration/operators/batch/generic.py

@@ -67,7 +67,9 @@ def execute(self, context) -> list[BatchIndexRow]:
         return rows
 
 
-class GeneticsBatchJobOperator(CloudBatchSubmitJobOperator):
+class BatchJobOperator(CloudBatchSubmitJobOperator):
+    """Generic Batch Job operator."""
+
     def __init__(
         self,
         job_name: str,

src/ot_orchestration/utils/batch.py

@@ -143,22 +143,25 @@ def create_batch_job(
     return job
 
 
-def create_task_env(var_list: list[dict[str, Any]]):
+def create_task_env(var_list: list[dict[str, str]]):
     """This function creates list of batch_v1.Environment objects from provided list of dictionaries."""
-    print(f"{var_list=}")
     environments = [Environment(variables=variables) for variables in var_list]
     return environments
 
 
 def create_task_commands(
-    commands: list[str] | None, params: dict[str, dict[str, Any] | None]
+    commands: list[str] | None, params: dict[str, str] | None
 ) -> list[str]:
     """This function prepares list of commands for google batch job from the step configuration."""
-    args = convert_params_to_hydra_positional_arg(params=params, dataproc=False)
     task_commands = []
+    args: list[str] = []
+    if params:
+        args = convert_params_to_hydra_positional_arg(params=params, dataproc=False)
     if commands:
         task_commands.extend(commands)
     task_commands.extend(args)
+    # ensure all are string values
+    task_commands = [str(t) for t in task_commands]
 
     if len(task_commands) > 1 and task_commands[0] == "-c":
         task_commands = ["-c", " ".join(task_commands[1:])]

src/ot_orchestration/utils/utils.py

@@ -138,7 +138,7 @@ def chain_dependencies(nodes: list[ConfigNode], tasks_or_task_groups: dict[str,
 
 
 def convert_params_to_hydra_positional_arg(
-    params: dict[str, Any] | None, dataproc: bool = False
+    params: dict[str, str] | None, dataproc: bool = False
 ) -> list[str]:
     """Convert configuration parameters to form that can be passed to hydra step positional arguments.