Merge pull request #299 from pyiron/dependencies

Define dependencies in submission script

pysqa/ext/modular.py

@@ -63,13 +63,14 @@ def submit_job(
             cores=cores,
             memory_max=memory_max,
             run_time_max=run_time_max,
+            dependency_list=dependency_list,
             command=command,
             **kwargs,
         )
         cluster_module = self._queue_to_cluster_dict[queue]
         commands = self._switch_cluster_command(
             cluster_module=cluster_module
-        ) + self._list_command_to_be_executed(dependency_list, queue_script_path)
+        ) + self._list_command_to_be_executed(queue_script_path=queue_script_path)
         out = self._execute_command(
             commands=commands,
             working_directory=working_directory,

pysqa/utils/basic.py

@@ -5,7 +5,7 @@
 import importlib
 import os
 import re
-from typing import Optional
+from typing import Optional, List
 
 import pandas
 from jinja2 import Template
@@ -178,7 +178,7 @@ def submit_job(
         )
         out = self._execute_command(
             commands=self._list_command_to_be_executed(
-                dependency_list, queue_script_path
+                queue_script_path=queue_script_path
             ),
             working_directory=working_directory,
             split_output=False,
@@ -188,14 +188,8 @@ def submit_job(
         else:
             return None
 
-    def _list_command_to_be_executed(
-        self, dependency_list: list[str], queue_script_path: str
-    ) -> list:
-        return (
-            self._commands.submit_job_command
-            + self._commands.dependencies(dependency_list)
-            + [queue_script_path]
-        )
+    def _list_command_to_be_executed(self, queue_script_path: str) -> list:
+        return self._commands.submit_job_command + [queue_script_path]
 
     def enable_reservation(self, process_id: int):
         """
@@ -354,6 +348,7 @@ def _write_queue_script(
         cores: Optional[int] = None,
         memory_max: Optional[int] = None,
         run_time_max: Optional[int] = None,
+        dependency_list: Optional[List[int]] = None,
         command: Optional[str] = None,
         **kwargs,
     ):
@@ -380,6 +375,7 @@ def _write_queue_script(
             cores=cores,
             memory_max=memory_max,
             run_time_max=run_time_max,
+            dependency_list=dependency_list,
             command=command,
             **kwargs,
         )
@@ -398,6 +394,7 @@ def _job_submission_template(
         cores: Optional[int] = None,
         memory_max: Optional[int] = None,
         run_time_max: Optional[int] = None,
+        dependency_list: Optional[List[int]] = None,
         command: Optional[str] = None,
         **kwargs,
     ) -> str:
@@ -410,6 +407,7 @@ def _job_submission_template(
             cores (int/None):
             memory_max (int/None):
             run_time_max (int/None):
+            dependency_list (list/None):
             command (str/None):
 
         Returns:
@@ -441,6 +439,7 @@ def _job_submission_template(
             memory_max=memory_max,
             run_time_max=run_time_max,
             command=command,
+            dependency_list=dependency_list,
             **kwargs,
         )
 

tests/config/flux/flux.sh

@@ -1,11 +1,16 @@
 #!/bin/bash
-# flux:--job-name={{job_name}}
+# flux: --job-name={{job_name}}
 # flux: --env=CORES={{cores}}
 # flux: --output=time.out
 # flux: --error=error.out
 # flux: -n {{cores}}
 {%- if run_time_max %}
 # flux: -t {{ [1, run_time_max // 60]|max }}
 {%- endif %}
+{%- if dependency %}
+{%- for jobid in dependency %}
+# flux: --dependency=afterok:{{jobid}}
+{%- endfor %}
+{%- endif %}
 
 {{command}}

tests/config/slurm/slurm.sh

@@ -7,6 +7,9 @@
 {%- if run_time_max %}
 #SBATCH --time={{ [1, run_time_max // 60]|max }}
 {%- endif %}
+{%- if dependency %}
+#SBATCH --dependency=afterok:{{ dependency | join(',') }}
+{%- endif %}
 {%- if memory_max %}
 #SBATCH --mem={{memory_max}}G
 {%- endif %}

tests/config/slurm/slurm_extra.sh

@@ -13,6 +13,9 @@
 {%- if run_time_max %}
 #SBATCH --time={{ [1, run_time_max // 60]|max }}
 {%- endif %}
+{%- if dependency %}
+#SBATCH --dependency=afterok:{{ dependency | join(',') }}
+{%- endif %}
 {%- if memory_max %}
 #SBATCH --mem={{memory_max}}G
 {%- endif %}

tests/test_flux.py

@@ -98,7 +98,7 @@ def execute_command(
             output = f.read()
         content = """\
 #!/bin/bash
-# flux:--job-name=test
+# flux: --job-name=test
 # flux: --env=CORES=4
 # flux: --output=time.out
 # flux: --error=error.out

tests/test_gent.py

@@ -56,12 +56,12 @@ def test_interfaces(self):
     def test__list_command_to_be_executed(self):
         with self.subTest("gent"):
             self.assertEqual(
-                self.gent._adapter._list_command_to_be_executed(None, "here"),
+                self.gent._adapter._list_command_to_be_executed("here"),
                 ["sbatch", "--parsable", "here"],
             )
         with self.subTest("gent with dependency"):
             self.assertRaises(
-                NotImplementedError,
+                TypeError,
                 self.gent._adapter._list_command_to_be_executed,
                 [],
                 "here",

tests/test_lsf.py

@@ -53,12 +53,12 @@ def test_interfaces(self):
     def test__list_command_to_be_executed(self):
         with self.subTest("lsf"):
             self.assertEqual(
-                self.lsf._adapter._list_command_to_be_executed(None, "here"),
+                self.lsf._adapter._list_command_to_be_executed("here"),
                 ["bsub", "here"],
             )
         with self.subTest("lsf with dependency"):
             self.assertRaises(
-                NotImplementedError,
+                TypeError,
                 self.lsf._adapter._list_command_to_be_executed,
                 [],
                 "here",

tests/test_moab.py

@@ -42,13 +42,13 @@ def test_interfaces(self):
     def test__list_command_to_be_executed(self):
         with self.subTest("moab with dependency"):
             self.assertRaises(
-                NotImplementedError,
+                TypeError,
                 self.moab._adapter._list_command_to_be_executed,
                 [],
                 "here",
             )
         with self.subTest("moab"):
             self.assertEqual(
-                self.moab._adapter._list_command_to_be_executed(None, "here"),
+                self.moab._adapter._list_command_to_be_executed("here"),
                 ["msub", "here"],
             )

tests/test_sge.py

@@ -110,12 +110,12 @@ def test_interfaces(self):
     def test__list_command_to_be_executed(self):
         with self.subTest("sge"):
             self.assertEqual(
-                self.sge._adapter._list_command_to_be_executed(None, "here"),
+                self.sge._adapter._list_command_to_be_executed("here"),
                 ["qsub", "-terse", "here"],
             )
         with self.subTest("sge with dependency"):
             self.assertRaises(
-                NotImplementedError,
+                TypeError,
                 self.sge._adapter._list_command_to_be_executed,
                 [],
                 "here",

tests/test_slurm.py

@@ -71,19 +71,9 @@ def test_interfaces(self):
     def test__list_command_to_be_executed(self):
         with self.subTest("slurm"):
             self.assertEqual(
-                self.slurm._adapter._list_command_to_be_executed(None, "here"),
+                self.slurm._adapter._list_command_to_be_executed("here"),
                 ["sbatch", "--parsable", "here"],
             )
-        with self.subTest("slurm with one dependency"):
-            self.assertEqual(
-                self.slurm._adapter._list_command_to_be_executed(["2"], "here"),
-                ["sbatch", "--parsable", "--dependency=afterok:2", "here"],
-            )
-        with self.subTest("slurm with two dependencies"):
-            self.assertEqual(
-                self.slurm._adapter._list_command_to_be_executed(["2", "34"], "here"),
-                ["sbatch", "--parsable", "--dependency=afterok:2,34", "here"],
-            )
 
     def test_convert_queue_status_slurm(self):
         with open(os.path.join(self.path, "config/slurm", "squeue_output"), "r") as f:

tests/test_torque.py

@@ -41,12 +41,12 @@ def test_interfaces(self):
     def test__list_command_to_be_executed(self):
         with self.subTest("torque"):
             self.assertEqual(
-                self.torque._adapter._list_command_to_be_executed(None, "here"),
+                self.torque._adapter._list_command_to_be_executed("here"),
                 ["qsub", "here"],
             )
         with self.subTest("torque with dependency"):
             self.assertRaises(
-                NotImplementedError,
+                TypeError,
                 self.torque._adapter._list_command_to_be_executed,
                 [],
                 "here",