support pipelinechannel on set env vars

Signed-off-by: JerT33 <trestjeremiah@gmail.com>

compiler updates

Signed-off-by: JerT33 <trestjeremiah@gmail.com>

Author: JerT33

backend/src/v2/driver/driver.go

@@ -267,7 +267,11 @@ func initPodSpecPatch(
 	// Convert environment variables
 	userEnvVar := make([]k8score.EnvVar, 0)
 	for _, envVar := range container.GetEnv() {
-		userEnvVar = append(userEnvVar, k8score.EnvVar{Name: envVar.GetName(), Value: envVar.GetValue()})
+		resolvedValue, err := resolvePodSpecInputRuntimeParameter(envVar.GetValue(), executorInput)
+		if err != nil {
+			return nil, fmt.Errorf("failed to resolve environment variable %q: %w", envVar.GetName(), err)
+		}
+		userEnvVar = append(userEnvVar, k8score.EnvVar{Name: envVar.GetName(), Value: resolvedValue})
 	}
 
 	userEnvVar = append(userEnvVar, proxy.GetConfig().GetEnvVars()...)

sdk/python/kfp/compiler/pipeline_spec_builder.py

@@ -145,6 +145,12 @@ def build_task_spec_for_task(
         if val and pipeline_channel.extract_pipeline_channels_from_any(val):
             task.inputs['base_image'] = val
 
+    if task.container_spec and task.container_spec.env:
+        for env_name, env_val in task.container_spec.env.items():
+            if env_val and pipeline_channel.extract_pipeline_channels_from_any(
+                    env_val):
+                task.inputs[env_name] = env_val
+
     for input_name, input_value in task.inputs.items():
         # Since LoopParameterArgument and LoopArtifactArgument and LoopArgumentVariable are narrower
         # types than PipelineParameterChannel, start with them.
@@ -748,7 +754,7 @@ def convert_to_placeholder(input_value: str) -> str:
             args=task.container_spec.args,
             env=[
                 pipeline_spec_pb2.PipelineDeploymentConfig.PipelineContainerSpec
-                .EnvVar(name=name, value=value)
+                .EnvVar(name=name, value=convert_to_placeholder(value))
                 for name, value in (task.container_spec.env or {}).items()
             ]))
 

sdk/python/kfp/dsl/pipeline_task.py

@@ -623,22 +623,43 @@ def set_display_name(self, name: str) -> 'PipelineTask':
         return self
 
     @block_if_final()
-    def set_env_variable(self, name: str, value: str) -> 'PipelineTask':
+    def set_env_variable(
+        self,
+        name: str,
+        value: Union[str, pipeline_channel.PipelineChannel],
+    ) -> 'PipelineTask':
         """Sets environment variable for the task.
 
         Args:
             name: Environment variable name.
-            value: Environment variable value.
+            value: Environment variable value. Supports dynamic values such as
+                Pipeline Parameters or outputs from previous tasks, which are
+                resolved at runtime.
 
         Returns:
             Self return to allow chained setting calls.
         """
         self._ensure_container_spec_exists()
 
+        pipeline_channels = (
+            pipeline_channel.extract_pipeline_channels_from_any(value))
+
+        if isinstance(value, pipeline_channel.PipelineChannel):
+            value = str(value)
+
         if self.container_spec.env is not None:
             self.container_spec.env[name] = value
         else:
             self.container_spec.env = {name: value}
+
+        if pipeline_channels:
+            existing_channel_patterns = {
+                channel.pattern for channel in self._channel_inputs
+            }
+            for channel in pipeline_channels:
+                if channel.pattern not in existing_channel_patterns:
+                    self._channel_inputs.append(channel)
+                    existing_channel_patterns.add(channel.pattern)
         return self
 
     @block_if_final()