initial commit of correct gdb states

beeflow/common/gdb/gdb_driver.py

@@ -159,6 +159,13 @@ def get_ready_tasks(self):
         :rtype: list of Task
         """
 
+    @abstractmethod
+    def get_no_start_tasks(self):
+        """Return tasks with state 'No Start' from the graph database.
+
+        :rtype: list of Task
+        """
+
     @abstractmethod
     def get_dependent_tasks(self, task):
         """Return the dependent tasks of a workflow task in the graph database.

beeflow/common/gdb/neo4j_cypher.py

@@ -421,6 +421,19 @@ def get_ready_tasks(tx, wf_id):
     return [rec['t'] for rec in tx.run(get_ready_query, wf_id=wf_id)]
 
 
+def get_no_start_tasks(tx, wf_id):
+    """Get all tasks that have the no start state.
+
+    :param workflow_id: the workflow id
+    :type workflow_id: str
+    :rtype: neo4j.Result
+    """
+    get_ready_query = ("MATCH (:Metadata {state: 'No Start'})-[:DESCRIBES]->"
+                       "(t:Task {workflow_id: $wf_id}) RETURN t")
+
+    return [rec['t'] for rec in tx.run(get_ready_query, wf_id=wf_id)]
+
+
 def get_dependent_tasks(tx, task):
     """Get the tasks that depend on a specified task.
 

beeflow/common/gdb/neo4j_driver.py

@@ -289,6 +289,17 @@ def get_ready_tasks(self, workflow_id):
         tuples = self._get_task_data_tuples(task_records)
         return [_reconstruct_task(tup[0], tup[1], tup[2], tup[3], tup[4]) for tup in tuples]
 
+    def get_no_start_tasks(self, workflow_id):
+        """Return tasks with state 'No Start' from the graph database from a particular workflow.
+
+        :param workflow_id: the workflow id
+        :type workflow_id: str
+        :rtype: list of Task
+        """
+        task_records = self._read_transaction(tx.get_no_start_tasks, wf_id=workflow_id)
+        tuples = self._get_task_data_tuples(task_records)
+        return [_reconstruct_task(tup[0], tup[1], tup[2], tup[3], tup[4]) for tup in tuples]
+
     def get_dependent_tasks(self, task):
         """Return the dependent tasks of a specified workflow task.
 

beeflow/common/wf_interface.py

@@ -200,6 +200,12 @@ def get_ready_tasks(self):
         """
         return self._gdb_driver.get_ready_tasks(self._workflow_id)
 
+    def set_no_start_tasks(self):
+        """Set no start tasks to waiting."""
+        tasks = self._gdb_driver.get_no_start_tasks(self._workflow_id)
+        for task in tasks:
+            self.set_task_state(task, "WAITING")
+
     def get_dependent_tasks(self, task):
         """Get the dependents of a task in a BEE workflow.
 

beeflow/wf_manager/resources/wf_utils.py

@@ -305,11 +305,7 @@ def start_workflow(wf_id):
     db = connect_db(wfm_db, get_db_path())
     wfi = get_workflow_interface(wf_id)
     state = wfi.get_workflow_state()
-    log.info(f"Leah state test: {state}")
-    if state in ('RUNNING', 'PAUSED', 'COMPLETED'):
-        return False
-    if state == 'No Start':
-
+    wfi.set_no_start_tasks()
     wfi.execute_workflow()
     tasks = wfi.get_ready_tasks()
     schedule_submit_tasks(wf_id, tasks)