serial: partial workflow execution

* Add target as operational option. * Addresses reanahub#54. Signed-off-by: Daniel Prelipcean <[email protected]>
dprelipcean · Sep 10, 2019 · 9112abd · 9112abd
1 parent 512173c
commit 9112abd
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 1 deletion.
diff --git a/reana_workflow_engine_serial/tasks.py b/reana_workflow_engine_serial/tasks.py
@@ -25,7 +25,8 @@
                     copy_workspace_to_cache, load_json, poll_job_status,
                     publish_cache_copy, publish_job_submission,
                     publish_job_success, publish_workflow_failure,
-                    publish_workflow_start, sanitize_command)
+                    publish_workflow_start, sanitize_command,
+                    validate_workflow)
 
 rjc_api_client = JobControllerAPIClient('reana-job-controller')
 
@@ -122,6 +123,9 @@ def run(workflow_json,
         publisher,
         cache_enabled):
     """Run a serial workflow."""
+    if operational_options:
+        workflow_json = validate_workflow(workflow_json, operational_options)
+
     expanded_workflow_json = serial_load(None,
                                          workflow_json,
                                          workflow_parameters)

diff --git a/reana_workflow_engine_serial/utils.py b/reana_workflow_engine_serial/utils.py
@@ -228,3 +228,25 @@ def publish_workflow_failure(job_id,
                                           build_progress_message(
                                               failed=failed_jobs)
                                       })
+
+
+def validate_workflow(workflow_json, operational_options):
+    """Validate the workflow given the operational options."""
+    target_step = operational_options.get('target')
+
+    if target_step:
+        # Target step can be either int or step name (str)
+        try:
+            step_number = int(target_step)
+        except ValueError:
+            # Get the step names from the workflow
+            input_steps = list()
+            for step in workflow_json['steps']:
+                input_steps.append(step['name'])
+
+            step_number = input_steps.index(target_step)
+
+        # Remove all subsequent steps
+        workflow_json['steps'] = workflow_json['steps'][:step_number+1]
+
+    return workflow_json