Send callback upon job start (#135)

* send callback upon job start * support file for "started' callback * revert sending callback from starting job * 'progress' callback in AsyncWorkflow * comment out unneded callback file reading in test (race condition) * Update nb2workflow/service.py --------- Co-authored-by: Volodymyr <[email protected]>
oda-hub · Jan 24, 2024 · 13af118 · 13af118
1 parent 2d5d9c1
commit 13af118
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 8 deletions.
diff --git a/nb2workflow/service.py b/nb2workflow/service.py
@@ -168,12 +168,12 @@ def _run(self):
             async_queue.put(self)
             app.async_workflows[self.key] = 'submitted'
             return
-
-        app.async_workflows[self.key] = 'started'
-
+
         template_nba = app.notebook_adapters.get(self.target)
-
         nba = NotebookAdapter(template_nba.notebook_fn, tempdir_cache=app.async_workflow_jobdirs)
+
+        app.async_workflows[self.key] = 'started'
+        self.perform_callback(action='progress')
 
         try:
             exceptions = nba.execute(self.params['request_parameters'], callback_url=self.callback, tmpdir_key=self.key)
@@ -223,7 +223,7 @@ def _run(self):
 
         self.perform_callback()
 
-    def perform_callback(self):
+    def perform_callback(self, action='done'):
         if self.callback is None:
             logger.info('no callback registered, skipping')
             return
@@ -234,7 +234,7 @@ def perform_callback(self):
         result = app.async_workflows[self.key]
 
         callback_payload = dict(
-            action='done'
+            action=action
         )
 
         if re.match('^file://', self.callback):

diff --git a/tests/test_service.py b/tests/test_service.py
@@ -125,8 +125,8 @@ def test_worker_run():
 
 
     while True:
-        if os.path.exists(callback_fn):
-            callback_json = json.load(open(callback_fn))
+        # if os.path.exists(callback_fn):
+        #     callback_json = json.load(open(callback_fn))
             # assert callback_json['action'] == 'done'
 
         options = client.get('/api/v1.0/options')