debug

src/engine/src/core/Server.py

@@ -39,7 +39,6 @@
 from utils import bytes_to_json, load_plugins, lbuffer_str as lbuf
 from errors import NoAvailableWorkers, WorkflowTerminated
 
-from pprint import pprint
 
 logger = logging.getLogger("server")
 
@@ -152,9 +151,8 @@ def _start_worker(self, body, connection, channel, delivery_tag):
         acked = False # Indicates that the message as been acked
         try:
             # Decode the message body, then convert to an object.
-            json_request = json.loads(bytes_to_json(body))
-            pprint(json_request)
-            request = WorkflowSubmissionRequest(**json_request)
+            deserialized_request = json.loads(bytes_to_json(body))
+            request = WorkflowSubmissionRequest(**deserialized_request)
 
             # Get a workflow executor worker. If there are none available,
             # this will raise a "NoWorkersAvailabe" error which is handled

src/engine/src/core/workflows/executors/WorkflowExecutor.py

@@ -435,10 +435,11 @@ def _set_tasks(self, tasks):
         # Add all tasks to the queue
         self.state.queue = [ task for task in self.state.tasks ]
 
-    @interceptable
+    @interceptable()
     def _prepare_pipeline(self):
         # Create all of the directories needed for the pipeline to run and persist results and cache
         self._prepare_pipeline_fs()
+        print(self.state.ctx)
 
         # template_mapper = TemplateMapper(cache_dir=self.state.ctx.pipeline.git_cache_dir)
         # if self.state.ctx.pipeline.uses != None: