feat: Env vars safe substitute (#188)

* feat: Added env vars to argo and safe substitute for config

* bug: in identifies

* bug: in identifies

examples/configs/argo-config.yaml

@@ -2,14 +2,19 @@ pipeline-executor:
   type: "argo" # (1)
   config:
     pvc_for_runnable: runnable
+    env:
+      - name: argo_env
+        value: "argo"
     argoWorkflow:
       metadata:
         generateName: "argo-" # (2)
         namespace: enterprise-mlops
       spec:
         serviceAccountName: "default-editor"
         templateDefaults:
-          image: harbor.csis.astrazeneca.net/mlops/runnable:latest # (2)
+          image: $docker_image # (2)
+
+
 
 
 run-log-store: # (4)

extensions/pipeline_executor/argo.py

@@ -244,6 +244,7 @@ class TemplateDefaults(BaseModelWIthConfig):
     image: str
     image_pull_policy: Optional[ImagePullPolicy] = Field(default=ImagePullPolicy.Always)
     resources: Resources = Field(default_factory=Resources)
+    env: list[EnvVar | SecretEnvVar] = Field(default_factory=list, exclude=True)
 
 
 # User provides this as part of the argoSpec
@@ -365,6 +366,7 @@ class ArgoExecutor(GenericPipelineExecutor):
     custom_volumes: Optional[list[CustomVolume]] = Field(
         default_factory=list[CustomVolume]
     )
+    env: list[EnvVar] = Field(default_factory=list[EnvVar])
 
     expose_parameters_as_inputs: bool = True
     secret_from_k8s: Optional[str] = Field(default=None)
@@ -508,6 +510,10 @@ def _create_container_template(
         )
 
         self._set_up_initial_container(container_template=core_container_template)
+        self._expose_secrets_to_task(
+            working_on=node, container_template=core_container_template
+        )
+        self._set_env_vars_to_task(node, core_container_template)
 
         container_template = ContainerTemplate(
             container=core_container_template,
@@ -523,12 +529,32 @@ def _create_container_template(
 
         return container_template
 
+    def _set_env_vars_to_task(
+        self, working_on: BaseNode, container_template: CoreContainerTemplate
+    ):
+        if not isinstance(working_on, TaskNode):
+            return
+        global_envs: dict[str, str] = {}
+
+        for env_var in self.env:
+            global_envs[env_var.name] = env_var.value
+
+        node_overide = {}
+        if hasattr(working_on, "overides"):
+            node_overide = working_on.overides
+
+        global_envs.update(node_overide.get("env", {}))
+        for key, value in global_envs.items():
+            env_var_to_add = EnvVar(name=key, value=value)
+            container_template.env.append(env_var_to_add)
+
     def _expose_secrets_to_task(
         self,
         working_on: BaseNode,
         container_template: CoreContainerTemplate,
     ):
-        assert isinstance(working_on, TaskNode)
+        if not isinstance(working_on, TaskNode):
+            return
         secrets = working_on.executable.secrets
         for secret in secrets:
             assert self.secret_from_k8s is not None

runnable/utils.py

@@ -109,8 +109,16 @@ def apply_variables(
         raise Exception("Argument Variables should be dict")
 
     json_d = json.dumps(apply_to)
-    transformed = str_template(json_d).substitute(**variables)
-    return json.loads(transformed)
+    string_template = str_template(json_d)
+
+    template = string_template.safe_substitute(variables)
+
+    if "$" in template:
+        logger.warning(
+            "Not all variables found in the config are found in the variables"
+        )
+
+    return json.loads(template)
 
 
 def get_module_and_attr_names(command: str) -> Tuple[str, str]:

tests/runnable/test_utils.py

@@ -4,10 +4,8 @@
 
 import pytest
 
-from runnable import (
-    defaults,  # pylint: disable=import-error
-    utils,  # pylint: disable=import-error
-)
+from runnable import defaults  # pylint: disable=import-error
+from runnable import utils  # pylint: disable=import-error
 
 
 def test_does_file_exist_returns_true_if_path_true(mocker, monkeypatch):
@@ -66,8 +64,8 @@ def test_apply_variables_applies_variables():
 def test_apply_variables_applies_known_variables():
     apply_to = "${var}_${var1}"
 
-    with pytest.raises(KeyError):
-        transformed = utils.apply_variables(apply_to, variables={"var": "hello"})
+    transformed = utils.apply_variables(apply_to, variables={"var": "hello"})
+    assert transformed == "hello_${var1}"
 
 
 def test_get_module_and_func_names_raises_exception_for_incorrect_command():