fix: Variable mapping now exposed as a Translation dataclass

Author: Alan Christie

workflow/decoder.py

@@ -4,6 +4,7 @@
 """
 
 import os
+from dataclasses import dataclass
 from typing import Any
 
 import jsonschema
@@ -23,6 +24,14 @@
 assert _WORKFLOW_SCHEMA
 
 
+@dataclass
+class Translation:
+    """A source ("in_") to destination ("out") variable map."""
+
+    in_: str
+    out: str
+
+
 def validate_schema(workflow: dict[str, Any]) -> str | None:
     """Checks the Workflow Definition against the built-in schema.
     If there's an error the error text is returned, otherwise None.
@@ -107,28 +116,27 @@ def get_step_input_variable_names(
     return variable_names
 
 
-def get_step_workflow_variable_mapping(
-    *, step: dict[str, Any]
-) -> list[tuple[str, str]]:
+def get_step_workflow_variable_mapping(*, step: dict[str, Any]) -> list[Translation]:
     """Returns a list of workflow vaiable name to step variable name tuples
     for the given step."""
-    variable_mapping: list[tuple[str, str]] = []
+    variable_mapping: list[Translation] = []
     if "variable-mapping" in step:
         for v_map in step["variable-mapping"]:
             if "from-workflow" in v_map:
-                # Tuple is "from" -> "to"
                 variable_mapping.append(
-                    (v_map["from-workflow"]["variable"], v_map["variable"])
+                    Translation(
+                        in_=v_map["from-workflow"]["variable"], out=v_map["variable"]
+                    )
                 )
     return variable_mapping
 
 
 def get_step_prior_step_variable_mapping(
     *, step: dict[str, Any]
-) -> dict[str, list[tuple[str, str]]]:
-    """Returns list of tuples, indexed by prior step name, of source step vaiable name
-    to this step's variable name."""
-    variable_mapping: dict[str, list[tuple[str, str]]] = {}
+) -> dict[str, list[Translation]]:
+    """Returns list of translate objects, indexed by prior step name,
+    that identify source step vaiable name to this step's variable name."""
+    variable_mapping: dict[str, list[Translation]] = {}
     if "variable-mapping" in step:
         for v_map in step["variable-mapping"]:
             if "from-step" in v_map:
@@ -137,10 +145,12 @@ def get_step_prior_step_variable_mapping(
                 # Tuple is "from" -> "to"
                 if step_name in variable_mapping:
                     variable_mapping[step_name].append(
-                        (step_variable, v_map["variable"])
+                        Translation(in_=step_variable, out=v_map["variable"])
                     )
                 else:
-                    variable_mapping[step_name] = [(step_variable, v_map["variable"])]
+                    variable_mapping[step_name] = [
+                        Translation(in_=step_variable, out=v_map["variable"])
+                    ]
     return variable_mapping
 
 

workflow/workflow_engine.py

@@ -39,6 +39,7 @@
 )
 
 from .decoder import (
+    Translation,
     get_step_input_variable_names,
     get_step_prior_step_variable_mapping,
     get_step_replicator,
@@ -364,14 +365,14 @@ 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 from_to in get_step_workflow_variable_mapping(step=step):
-            all_variables[from_to[1]] = running_workflow_variables[from_to[0]]
+        for tr in get_step_workflow_variable_mapping(step=step):
+            all_variables[tr.out] = running_workflow_variables[tr.in_]
 
         # Now we apply variables from the "variable mapping" block
         # related to values used in prior steps. The decoder gives
         # us a map indexed by prior step name that's a list of "in" "out"
         # tuples as above.
-        step_prior_v_map: dict[str, list[tuple[str, str]]] = (
+        step_prior_v_map: dict[str, list[Translation]] = (
             get_step_prior_step_variable_mapping(step=step)
         )
         for prior_step_name, v_map in step_prior_v_map.items():
@@ -381,8 +382,8 @@ def _validate_step_command(
                 name=prior_step_name, running_workflow_id=running_workflow_id
             )
             # Copy "in" value to "out"...
-            for from_to in v_map:
-                all_variables[from_to[1]] = prior_step["variables"][from_to[0]]
+            for tr in v_map:
+                all_variables[tr.out] = prior_step["variables"][tr.in_]
 
         _LOGGER.debug(
             "Index %s (%s) workflow_variables=%s",

workflow/workflow_validator.py

@@ -123,17 +123,17 @@ def _validate_tag_level(
                 .get("variable")
             ):
                 found: bool = False
-                for variable_map in get_step_workflow_variable_mapping(step=step):
-                    if replicate_using_input == variable_map[0]:
+                for translation in get_step_workflow_variable_mapping(step=step):
+                    if replicate_using_input == translation.out:
                         found = True
                         break
                 if not found:
                     for (
                         step_name,
-                        variable_map_list,
+                        translations,
                     ) in get_step_prior_step_variable_mapping(step=step).items():
-                        for variable_map in variable_map_list:
-                            if replicate_using_input == variable_map[0]:
+                        for translation in translations:
+                            if replicate_using_input == translation.out:
                                 found = True
                                 break
                         if found: