fix: Major refactoring of logic (for new launch/workflow API)

Author: Alan Christie

tests/instance_launcher.py

@@ -68,18 +68,32 @@ def __init__(
                 elif os.path.isdir(file_path):
                     shutil.rmtree(file_path)
 
-    def launch(self, launch_parameters: LaunchParameters) -> LaunchResult:
+    def launch(self, *, launch_parameters: LaunchParameters) -> LaunchResult:
         assert launch_parameters
         assert launch_parameters.project_id == TEST_PROJECT_ID
         assert launch_parameters.specification
         assert isinstance(launch_parameters.specification, dict)
 
         os.makedirs(EXECUTION_DIRECTORY, exist_ok=True)
 
-        # Create an Instance record (and dummy Task ID)
-        response = self._api_adapter.create_instance(
-            running_workflow_step_id=launch_parameters.running_workflow_step_id
+        # Create a running workflow step
+        assert launch_parameters.running_workflow_id
+        assert launch_parameters.step_name
+        response, _ = self._api_adapter.create_running_workflow_step(
+            running_workflow_id=launch_parameters.running_workflow_id,
+            step=launch_parameters.step_name,
+            replica=launch_parameters.step_replication_number,
         )
+        assert "id" in response
+        rwfs_id: str = response["id"]
+        # And add the variables we've been provided with
+        if launch_parameters.variables:
+            _ = self._api_adapter.set_running_workflow_step_variables(
+                running_workflow_step_id=rwfs_id, variables=launch_parameters.variables
+            )
+
+        # Create an Instance record (and dummy Task ID)
+        response = self._api_adapter.create_instance(running_workflow_step_id=rwfs_id)
         instance_id = response["id"]
         task_id = "task-00000000-0000-0000-0000-000000000001"
 
@@ -96,8 +110,8 @@ def launch(self, launch_parameters: LaunchParameters) -> LaunchResult:
         # The command may not need any, but we do the decoding anyway.
         decoded_command, status = job_decoder.decode(
             job["command"],
-            launch_parameters.specification_variables,
-            launch_parameters.running_workflow_step_id,
+            launch_parameters.variables,
+            rwfs_id,
             TextEncoding.JINJA2_3_0,
         )
         print(f"Decoded command: {decoded_command}")
@@ -129,6 +143,7 @@ def launch(self, launch_parameters: LaunchParameters) -> LaunchResult:
         self._msg_dispatcher.send(pod_message)
 
         return LaunchResult(
+            running_workflow_step_id=rwfs_id,
             instance_id=instance_id,
             task_id=task_id,
             command=" ".join(subprocess_cmd),

tests/test_test_instance_launcher.py

@@ -33,23 +33,18 @@ def test_launch_nop(basic_launcher):
         project_id=TEST_PROJECT_ID,
         variables={},
     )
-    response, _ = utaa.create_running_workflow_step(
-        running_workflow_id=response["id"], step="step-1"
-    )
-    rwfsid = response["id"]
     lp: LaunchParameters = LaunchParameters(
         project_id=TEST_PROJECT_ID,
         name="Test Instance",
         launching_user_name="dlister",
         launching_user_api_token="1234567890",
         running_workflow_id=rwfid,
-        running_workflow_step_id=rwfsid,
+        step_name="step-1",
         specification={"collection": "workflow-engine-unit-test-jobs", "job": "nop"},
-        specification_variables={},
     )
 
     # Act
-    result = launcher.launch(lp)
+    result = launcher.launch(launch_parameters=lp)
 
     # Assert
     assert result.error_num == 0
@@ -69,26 +64,21 @@ def test_launch_nop_fail(basic_launcher):
         variables={},
     )
     rwfid = response["id"]
-    response, _ = utaa.create_running_workflow_step(
-        running_workflow_id=response["id"], step="step-1"
-    )
-    rwfsid = response["id"]
     lp: LaunchParameters = LaunchParameters(
         project_id=TEST_PROJECT_ID,
         name="Test Instance",
         launching_user_name="dlister",
         launching_user_api_token="1234567890",
         running_workflow_id=rwfid,
-        running_workflow_step_id=rwfsid,
+        step_name="step-1",
         specification={
             "collection": "workflow-engine-unit-test-jobs",
             "job": "nop-fail",
         },
-        specification_variables={},
     )
 
     # Act
-    result = launcher.launch(lp)
+    result = launcher.launch(launch_parameters=lp)
 
     # Assert
     assert result.error_num == 0
@@ -108,26 +98,22 @@ def test_launch_smiles_to_file(basic_launcher):
         variables={},
     )
     rwfid = response["id"]
-    response, _ = utaa.create_running_workflow_step(
-        running_workflow_id=response["id"], step="step-1"
-    )
-    rwfsid = response["id"]
     lp: LaunchParameters = LaunchParameters(
         project_id=TEST_PROJECT_ID,
         name="Test Instance",
         launching_user_name="dlister",
         launching_user_api_token="1234567890",
         running_workflow_id=rwfid,
-        running_workflow_step_id=rwfsid,
+        step_name="step-1",
         specification={
             "collection": "workflow-engine-unit-test-jobs",
             "job": "smiles-to-file",
         },
-        specification_variables={"smiles": "C1=CC=CC=C1", "outputFile": "output.smi"},
+        variables={"smiles": "C1=CC=CC=C1", "outputFile": "output.smi"},
     )
 
     # Act
-    result = launcher.launch(lp)
+    result = launcher.launch(launch_parameters=lp)
 
     # Assert
     assert result.error_num == 0

tests/test_workflow_engine_examples.py

@@ -398,6 +398,7 @@ def test_workflow_engine_simple_python_molprops_with_options(basic_engine):
     assert project_file_exists(output_file_2)
 
 
+@pytest.mark.skip(reason="WIP")
 def test_workflow_engine_simple_python_fanout(basic_engine):
     # Arrange
     md, da = basic_engine

tests/workflow-definitions/simple-python-fanout.yaml

@@ -13,8 +13,9 @@ steps:
     job: splitsmiles
     version: "1.0.0"
     variables:
-      name: "count"
+      name: count
       value: "1"
+      outputFile: results.smi
   variable-mapping:
   - variable: inputFile
     from-workflow:
@@ -27,8 +28,9 @@ steps:
     job: append-col
     version: "1.0.0"
     variables:
-      name: "desc1"
+      name: desc1
       value: "777"
+      outputFile: results.smi
   replicate:
     using:
       variable: inputFile

workflow/workflow_abc.py

@@ -27,36 +27,23 @@ class LaunchParameters:
     specification: dict[str, Any]
     # An alternative way to pass variables to the specification.
     # If used it will replace any 'variables' already present in the specification.
-    specification_variables: dict[str, Any] | None = None
+    variables: dict[str, Any] | None = None
     # A string. In DM v4 converted to a boolean and set in the
     # instance Pod as a label. Setting this means the Instances
     # that are created will not be automatically removed by the Job operator.
     debug: str | None = None
     # The RunningWorkflow UUID.
     # Required if the Instance is part of a Workflow step.
     running_workflow_id: str | None = None
-    # The RunningWorkflowStep UUID.
+    # The RunningWorkflow's step name.
     # Required if the Instance is part of a Workflow step.
-    running_workflow_step_id: str | None = None
-    # A list of prior workflow steps that this step depends upon.
-    #
-    # This list gives the InstanceLauncher an opportunity to take the outputs
-    # of a prior instance and link them to the instance directory for the
-    # instance to be launched. We need to do this for Workflows because Instances
-    # run as apart of a Workflow do not automatically have their outputs copied (linked)
-    # to the Project directory when they complete. As an example, a step that relies
-    # on the output files from two prior steps will provide the following list: -
-    #
-    #   ["r-workflow-step-a04d", "r-workflow-step-d904"]
-    running_workflow_step_prior_steps: list[str] | None = None
-    # Workflow step Job inputs (for this step Instance). These Workflow Inputs (files)
-    # are a list of Job input variable names for file variables where the
-    # file is expected to be present in the Project directory. It is simply a list of
-    # Job variable names. The launcher is expected to find the 'value' of these
-    # variables and then move the file to the instance directory.
-    #
-    #   ["inputFile"]
-    running_workflow_step_inputs: list[str] | None = None
+    step_name: str | None = None
+    # The step replication number.
+    # If only one instance of the step is expected to run
+    # this value can be left at 0 (zero). If this step's launch
+    # is expected to be executed more than once the value should be
+    # non-zero (and unique for this workflow run).
+    step_replication_number: int = 0
     # The application ID (a custom resource name)
     # used to identify the 'type' of Instance to create.
     # For DM Jobs this will be 'datamanagerjobs.squonk.it'
@@ -75,6 +62,9 @@ class LaunchResult:
     # The following optional properties
     # may not be present if there's a launch error.
     #
+    # A running workflow step UUID
+    # (if the step is part of a running workflow)
+    running_workflow_step_id: str | None = None
     # The Instance UUID that was created for you.
     instance_id: str | None = None
     # The Task UUID that is handling the Instance launch
@@ -94,7 +84,6 @@ def launch(
         self,
         *,
         launch_parameters: LaunchParameters,
-        **kwargs: str,
     ) -> LaunchResult:
         """Launch a (Job) Instance"""
 
@@ -199,25 +188,6 @@ def set_running_workflow_done(
         """Set the success value for a RunningWorkflow Record.
         If not successful an error code and message should be provided."""
 
-    @abstractmethod
-    def create_running_workflow_step(
-        self,
-        *,
-        running_workflow_id: str,
-        step: str,
-        replica: int = 0,
-        prior_running_workflow_step_id: str | None = None,
-    ) -> tuple[dict[str, Any], int]:
-        """Create a RunningWorkflowStep Record (from a RunningWorkflow).
-        If this is a replica (concurrent execution) of a step the replica
-        value must be set to a value greater than 0. The replica is unique
-        for a given step and is used to distinguish between running workflow steps
-        generated from the same step name."""
-        # Should return:
-        # {
-        #    "id": "r-workflow-step-00000000-0000-0000-0000-000000000001",
-        # }
-
     @abstractmethod
     def get_running_workflow_step(
         self, *, running_workflow_step_id: str
@@ -292,17 +262,6 @@ def get_running_workflow_step_by_name(
         #          "id": "r-workflow-step-00000000-0000-0000-0000-000000000001",
         #       },
 
-    @abstractmethod
-    def set_running_workflow_step_variables(
-        self,
-        *,
-        running_workflow_step_id: str,
-        variables: dict[str, Any],
-    ) -> None:
-        """Set the variables used prior to decoding the step command for each step.
-        This can be used to understand step failures but will also be vital
-        when adding variables values to subsequent steps from prior step values."""
-
     @abstractmethod
     def set_running_workflow_step_done(
         self,