fix #2802: Set ImagePullPolicy per pipeline.

sdk/python/kfp/compiler/compiler.py

@@ -686,6 +686,12 @@ def _create_pipeline_workflow(self, args, pipeline, op_transformers=None, pipeli
 
     if exit_handler:
       workflow['spec']['onExit'] = exit_handler.name
+
+    if pipeline_conf.image_pull_policy != None and pipeline_conf.image_pull_policy in ["Always", "Never", "IfNotPresent"]:
+      for template in workflow["spec"]["templates"]:
+        if "container" in template:
+          template["container"]["imagePullPolicy"] = pipeline_conf.image_pull_policy
+
     return workflow
 
   def _validate_exit_handler(self, pipeline):

sdk/python/kfp/dsl/_pipeline.py

@@ -62,6 +62,7 @@ def __init__(self):
     self.ttl_seconds_after_finished = -1
     self.artifact_location = None
     self.op_transformers = []
+    self.image_pull_policy = None
 
   def set_image_pull_secrets(self, image_pull_secrets):
     """Configures the pipeline level imagepullsecret
@@ -126,6 +127,16 @@ def foo_pipeline(tag: str, pull_image_policy: str):
     self.artifact_location = artifact_location
     return self
 
+  def set_image_pull_policy(self, policy: str):
+    """Configures the default image pull policy
+
+    Args:
+      policy: the pull policy, has to be one of: Always, Never, IfNotPresent. 
+      For more info: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Container.md
+    """
+    self.image_pull_policy = policy
+    return self
+
   def add_op_transformer(self, transformer):
     """Configures the op_transformers which will be applied to all ops in the pipeline.
 
@@ -254,4 +265,3 @@ def _set_metadata(self, metadata):
     '''
     self._metadata = metadata
 
-

sdk/python/tests/compiler/compiler_tests.py

@@ -682,6 +682,28 @@ def some_pipeline():
       container = template.get('container', None)
       if container:
         self.assertEqual(template['retryStrategy']['limit'], 5)
+
+  def test_image_pull_policy(self):
+    def some_op():
+      return dsl.ContainerOp(
+          name='sleep',
+          image='busybox',
+          command=['sleep 1'],
+      )
+
+    @dsl.pipeline(name='some_pipeline')
+    def some_pipeline():
+      task1 = some_op()
+      task2 = some_op()
+      task3 = some_op()
+
+      dsl.get_pipeline_conf().set_image_pull_policy(policy="Allways")
+      workflow_dict = compiler.Compiler()._compile(some_pipeline)
+      for template in workflow_dict['spec']['templates']:
+        container = template.get('container', None)
+        if container:
+          self.assertEqual(template['container']['imagePullPolicy'], "Allways")
+
 
   def test_container_op_output_error_when_no_or_multiple_outputs(self):