refactor: Better function and variable naming (plumbing)

Author: Alan Christie

workflow/decoder.py

@@ -127,38 +127,42 @@ def get_step_input_variable_names(
     return variable_names
 
 
-def get_step_workflow_plumbing(*, step: dict[str, Any]) -> list[Connector]:
-    """Returns a list of workflow vaiable name to step variable name
-    Translation objects for the given step."""
-    variable_mapping: list[Connector] = []
-    if "plumbing" in step:
-        for v_map in step["plumbing"]:
+def get_step_workflow_variable_connections(
+    *, step_definition: dict[str, Any]
+) -> list[Connector]:
+    """Returns a list of connectors that connect a workflow variable name
+    to a step variable name for the given step definition."""
+    connections: list[Connector] = []
+    if "plumbing" in step_definition:
+        for v_map in step_definition["plumbing"]:
             if "from-workflow" in v_map:
-                variable_mapping.append(
+                connections.append(
                     Connector(
                         in_=v_map["from-workflow"]["variable"], out=v_map["variable"]
                     )
                 )
-    return variable_mapping
-
-
-def get_step_prior_step_plumbing(*, step: dict[str, Any]) -> dict[str, list[Connector]]:
-    """Returns list of Translation objects, indexed by prior step name,
-    that identify source step (output) variable name to this step's (input)
-    variable name."""
-    variable_mapping: dict[str, list[Connector]] = {}
-    if "plumbing" in step:
-        for v_map in step["plumbing"]:
+    return connections
+
+
+def get_step_prior_step_plumbing(
+    *, step_definition: dict[str, Any]
+) -> dict[str, list[Connector]]:
+    """Returns list of variable Connections, indexed by prior step name,
+    that identify a source step variable name (an output) to an input variable in this
+    step (an input)."""
+    plumbing: dict[str, list[Connector]] = {}
+    if "plumbing" in step_definition:
+        for v_map in step_definition["plumbing"]:
             if "from-step" in v_map:
                 step_name = v_map["from-step"]["name"]
                 step_variable = v_map["from-step"]["variable"]
                 # Tuple is "from" -> "to"
-                if step_name in variable_mapping:
-                    variable_mapping[step_name].append(
+                if step_name in plumbing:
+                    plumbing[step_name].append(
                         Connector(in_=step_variable, out=v_map["variable"])
                     )
                 else:
-                    variable_mapping[step_name] = [
+                    plumbing[step_name] = [
                         Connector(in_=step_variable, out=v_map["variable"])
                     ]
-    return variable_mapping
+    return plumbing

workflow/workflow_engine.py

@@ -43,7 +43,7 @@
     Connector,
     get_step,
     get_step_prior_step_plumbing,
-    get_step_workflow_plumbing,
+    get_step_workflow_variable_connections,
 )
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -129,7 +129,7 @@ def _handle_workflow_start_message(self, r_wfid: str) -> None:
         # Launch it.
         # If there's a launch problem the step (and running workflow) will have
         # and error, stopping it. There will be no Pod event as the launch has failed.
-        self._launch(wf=wf_response, rwf=rwf_response, step=first_step)
+        self._launch(wf=wf_response, rwf=rwf_response, step_definition=first_step)
 
     def _handle_workflow_stop_message(self, r_wfid: str) -> None:
         """Logic to handle a STOP message."""
@@ -265,7 +265,9 @@ def _handle_pod_message(self, msg: PodMessage) -> None:
                     # There's another step!
                     # For this simple logic it is the next step.
                     next_step = wf_response["steps"][step_index + 1]
-                    self._launch(wf=wf_response, rwf=rwf_response, step=next_step)
+                    self._launch(
+                        wf=wf_response, rwf=rwf_response, step_definition=next_step
+                    )
 
                     # Something was started (or there was a launch error and the step
                     # and running workflow error will have been set).
@@ -330,7 +332,7 @@ def _validate_step_command(
         # "in" variables are worklfow variables, and "out" variables
         # are expected Job variables. We use this to add variables
         # to the "all variables" map.
-        for connector in get_step_workflow_plumbing(step=step):
+        for connector in get_step_workflow_variable_connections(step_definition=step):
             assert connector.in_ in running_workflow_variables
             all_variables[connector.out] = running_workflow_variables[connector.in_]
 
@@ -339,7 +341,7 @@ def _validate_step_command(
         # us a map indexed by prior step name that's a list of "in" "out"
         # tuples as above.
         prior_step_plumbing: dict[str, list[Connector]] = get_step_prior_step_plumbing(
-            step=step
+            step_definition=step
         )
         for prior_step_name, connections in prior_step_plumbing.items():
             # Retrieve the prior "running" step
@@ -360,9 +362,13 @@ def _validate_step_command(
         return all_variables if success else message
 
     def _launch(
-        self, *, wf: dict[str, Any], rwf: dict[str, Any], step: dict[str, Any]
+        self,
+        *,
+        wf: dict[str, Any],
+        rwf: dict[str, Any],
+        step_definition: dict[str, Any],
     ) -> None:
-        step_name: str = step["name"]
+        step_name: str = step_definition["name"]
         rwf_id: str = rwf["id"]
         project_id = rwf["project"]["id"]
 
@@ -376,7 +382,7 @@ def _launch(
         rwf_variables: dict[str, Any] = rwf.get("variables", {})
         error_or_variables: str | dict[str, Any] = self._validate_step_command(
             running_workflow_id=rwf_id,
-            step=step,
+            step=step_definition,
             running_workflow_variables=rwf_variables,
         )
         if isinstance(error_or_variables, str):
@@ -401,7 +407,9 @@ def _launch(
         # be more than one prior step variable that is 'files'!
         replication_values: list[str] = []
         iter_variable: str | None = None
-        plumbing: dict[str, list[Connector]] = get_step_prior_step_plumbing(step=step)
+        plumbing: dict[str, list[Connector]] = get_step_prior_step_plumbing(
+            step_definition=step_definition
+        )
         for p_step_name, connections in plumbing.items():
             # We need to get the Job definition for each step
             # and then check whether the (ouptu) variable is of type 'files'...
@@ -468,7 +476,7 @@ def _launch(
                 debug=rwf.get("debug"),
                 launching_user_name=rwf["running_user"],
                 launching_user_api_token=rwf["running_user_api_token"],
-                specification=step["specification"],
+                specification=step_definition["specification"],
                 variables=variables,
                 running_workflow_id=rwf_id,
                 step_name=step_name,