We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
mount_pvc failing when pipeline includes ResourceOps
### What steps did you take:

Created a persistent volume claim using a `VolumeOp`:

```python
    vop = dsl.VolumeOp(
        name="shared-data",
        resource_name="shared-data",
        size="100G",
        storage_class="nfs",
    ).set_display_name("Prepare volume")
```

Attempted to mount the PVC to every operation in a pipeline using:

```python
    dsl.get_pipeline_conf().add_op_transformer(
        mount_pvc(pvc_name=vop.volume.persistent_volume_claim.claim_name, volume_mount_path=MOUNT_PATH)
    )
```

### What happened:
During compilation, the `add_op_transformer` attempts to invoke the `add_volume_mount` method on the `VolumeOp`, producing an error:

```
AttributeError: 'VolumeOp' object has no attribute 'add_volume_mount'
```

### What did you expect to happen:

The documentation states that transfomers take and return a `ContainerOp`:

https://kubeflow-pipelines.readthedocs.io/en/latest/source/kfp.dsl.html#kfp.dsl.PipelineConf.add_op_transformer

Given that, I would expect them to be bypassed for `VolumeOp` (or other `ResourceOp` operations), and hence this compilation should succeed.

### Environment:
<!-- Please fill in those that seem relevant. -->

KFP version: Build commit: 9c16e12

KFP SDK version: <!-- Please attach the output of this shell command: $pip list | grep kfp -->
kfp                      0.5.1
kfp-server-api           0.3.0

### Anything else you would like to add:
[Miscellaneous information that will assist in solving the issue.]

/kind bug
<!-- Please include labels by uncommenting them to help us better triage issues, choose from the following -->
<!--
// /area frontend
// /area backend
// /area sdk
// /area testing
// /area engprod
-->


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/kfp/dsl/_pipeline.py`
Content:
```
1 # Copyright 2018-2019 Google LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 
16 from . import _container_op
17 from . import _resource_op
18 from . import _ops_group
19 from ._component_bridge import _create_container_op_from_component_and_arguments
20 from ..components import _components
21 from ..components._naming import _make_name_unique_by_adding_index
22 import sys
23 
24 
25 # This handler is called whenever the @pipeline decorator is applied.
26 # It can be used by command-line DSL compiler to inject code that runs for every pipeline definition.
27 _pipeline_decorator_handler = None
28 
29 
30 def pipeline(name : str = None, description : str = None):
31   """Decorator of pipeline functions.
32 
33   Usage:
34   ```python
35   @pipeline(
36     name='my awesome pipeline',
37     description='Is it really awesome?'
38   )
39   def my_pipeline(a: PipelineParam, b: PipelineParam):
40     ...
41   ```
42   """
43   def _pipeline(func):
44     if name:
45       func._component_human_name = name
46     if description:
47       func._component_description = description
48 
49     if _pipeline_decorator_handler:
50       return _pipeline_decorator_handler(func) or func
51     else:
52       return func
53 
54   return _pipeline
55 
56 class PipelineConf():
57   """PipelineConf contains pipeline level settings
58   """
59   def __init__(self):
60     self.image_pull_secrets = []
61     self.timeout = 0
62     self.ttl_seconds_after_finished = -1
63     self.op_transformers = []
64     self.default_pod_node_selector = {}
65     self.image_pull_policy = None
66     self.parallelism = None
67 
68   def set_image_pull_secrets(self, image_pull_secrets):
69     """Configures the pipeline level imagepullsecret
70 
71     Args:
72       image_pull_secrets: a list of Kubernetes V1LocalObjectReference
73       For detailed description, check Kubernetes V1LocalObjectReference definition
74       https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1LocalObjectReference.md
75     """
76     self.image_pull_secrets = image_pull_secrets
77     return self
78 
79   def set_timeout(self, seconds: int):
80     """Configures the pipeline level timeout
81 
82     Args:
83       seconds: number of seconds for timeout
84     """
85     self.timeout = seconds
86     return self
87 
88   def set_parallelism(self, max_num_pods: int):
89     """Configures the max number of total parallel pods that can execute at the same time in a workflow.
90 
91     Args:
92         max_num_pods (int): max number of total parallel pods.
93     """
94     self.parallelism = max_num_pods
95     return self
96 
97   def set_ttl_seconds_after_finished(self, seconds: int):
98     """Configures the ttl after the pipeline has finished.
99 
100     Args:
101       seconds: number of seconds for the workflow to be garbage collected after it is finished.
102     """
103     self.ttl_seconds_after_finished = seconds
104     return self
105   
106   def set_default_pod_node_selector(self, label_name: str, value: str): 
107     """Add a constraint for nodeSelector for a pipeline. Each constraint is a key-value pair label. For the 
108       container to be eligible to run on a node, the node must have each of the constraints appeared
109       as labels.
110 
111     Args:
112         label_name: The name of the constraint label.
113         value: The value of the constraint label.
114     """
115     self.default_pod_node_selector[label_name] = value
116     return self
117   
118 
119   def set_image_pull_policy(self, policy: str):
120     """Configures the default image pull policy
121 
122     Args:
123       policy: the pull policy, has to be one of: Always, Never, IfNotPresent.
124       For more info: https://github.com/kubernetes-client/python/blob/10a7f95435c0b94a6d949ba98375f8cc85a70e5a/kubernetes/docs/V1Container.md
125     """
126     self.image_pull_policy = policy
127     return self
128 
129   def add_op_transformer(self, transformer):
130     """Configures the op_transformers which will be applied to all ops in the pipeline.
131 
132     Args:
133       transformer: a function that takes a ContainOp as input and returns a ContainerOp
134     """
135     self.op_transformers.append(transformer)
136 
137 
138 def get_pipeline_conf():
139   """Configure the pipeline level setting to the current pipeline
140     Note: call the function inside the user defined pipeline function.
141   """
142   return Pipeline.get_default_pipeline().conf
143 
144 #TODO: Pipeline is in fact an opsgroup, refactor the code.
145 class Pipeline():
146   """A pipeline contains a list of operators.
147 
148   This class is not supposed to be used by pipeline authors since pipeline authors can use
149   pipeline functions (decorated with @pipeline) to reference their pipelines. This class
150   is useful for implementing a compiler. For example, the compiler can use the following
151   to get the pipeline object and its ops:
152 
153   ```python
154   with Pipeline() as p:
155     pipeline_func(*args_list)
156 
157   traverse(p.ops)
158   ```
159   """
160 
161   # _default_pipeline is set when it (usually a compiler) runs "with Pipeline()"
162   _default_pipeline = None
163 
164   @staticmethod
165   def get_default_pipeline():
166     """Get default pipeline. """
167     return Pipeline._default_pipeline
168 
169   @staticmethod
170   def add_pipeline(name, description, func):
171     """Add a pipeline function with the specified name and description."""
172     # Applying the @pipeline decorator to the pipeline function
173     func = pipeline(name=name, description=description)(func)
174 
175   def __init__(self, name: str):
176     """Create a new instance of Pipeline.
177 
178     Args:
179       name: the name of the pipeline. Once deployed, the name will show up in Pipeline System UI.
180     """
181     self.name = name
182     self.ops = {}
183     # Add the root group.
184     self.groups = [_ops_group.OpsGroup('pipeline', name=name)]
185     self.group_id = 0
186     self.conf = PipelineConf()
187     self._metadata = None
188 
189   def __enter__(self):
190     if Pipeline._default_pipeline:
191       raise Exception('Nested pipelines are not allowed.')
192 
193     Pipeline._default_pipeline = self
194     self._old_container_task_constructor = _components._container_task_constructor
195     _components._container_task_constructor = _create_container_op_from_component_and_arguments
196 
197     def register_op_and_generate_id(op):
198       return self.add_op(op, op.is_exit_handler)
199 
200     self._old__register_op_handler = _container_op._register_op_handler
201     _container_op._register_op_handler = register_op_and_generate_id
202     return self
203 
204   def __exit__(self, *args):
205     Pipeline._default_pipeline = None
206     _container_op._register_op_handler = self._old__register_op_handler
207     _components._container_task_constructor = self._old_container_task_constructor
208 
209   def add_op(self, op: _container_op.BaseOp, define_only: bool):
210     """Add a new operator.
211 
212     Args:
213       op: An operator of ContainerOp, ResourceOp or their inherited types.
214 
215     Returns
216       op_name: a unique op name.
217     """
218     #If there is an existing op with this name then generate a new name.
219     op_name = _make_name_unique_by_adding_index(op.human_name, list(self.ops.keys()), ' ')
220 
221     self.ops[op_name] = op
222     if not define_only:
223       self.groups[-1].ops.append(op)
224 
225     return op_name
226 
227   def push_ops_group(self, group: _ops_group.OpsGroup):
228     """Push an OpsGroup into the stack.
229 
230     Args:
231       group: An OpsGroup. Typically it is one of ExitHandler, Branch, and Loop.
232     """
233     self.groups[-1].groups.append(group)
234     self.groups.append(group)
235 
236   def pop_ops_group(self):
237     """Remove the current OpsGroup from the stack."""
238     del self.groups[-1]
239 
240   def remove_op_from_groups(self, op):
241     for group in self.groups:
242       group.remove_op_recursive(op)
243 
244   def get_next_group_id(self):
245     """Get next id for a new group. """
246 
247     self.group_id += 1
248     return self.group_id
249 
250   def _set_metadata(self, metadata):
251     '''_set_metadata passes the containerop the metadata information
252     Args:
253       metadata (ComponentMeta): component metadata
254     '''
255     self._metadata = metadata
256
```

Path: `sdk/python/kfp/onprem.py`
Content:
```
1 
2 def mount_pvc(pvc_name='pipeline-claim', volume_name='pipeline', volume_mount_path='/mnt/pipeline'):
3     """
4         Modifier function to apply to a Container Op to simplify volume, volume mount addition and 
5         enable better reuse of volumes, volume claims across container ops.
6         Usage:
7             train = train_op(...)
8             train.apply(mount_pvc('claim-name', 'pipeline', '/mnt/pipeline'))
9     """
10     def _mount_pvc(task):
11         from kubernetes import client as k8s_client
12         local_pvc = k8s_client.V1PersistentVolumeClaimVolumeSource(claim_name=pvc_name)
13         return (
14             task
15                 .add_volume(
16                     k8s_client.V1Volume(name=volume_name, persistent_volume_claim=local_pvc)
17                 )
18                 .add_volume_mount(
19                     k8s_client.V1VolumeMount(mount_path=volume_mount_path, name=volume_name)
20                 )
21         )
22     return _mount_pvc
23
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sdk/python/kfp/dsl/_pipeline.py b/sdk/python/kfp/dsl/_pipeline.py
--- a/sdk/python/kfp/dsl/_pipeline.py
+++ b/sdk/python/kfp/dsl/_pipeline.py
@@ -102,9 +102,9 @@
     """
     self.ttl_seconds_after_finished = seconds
     return self
-  
-  def set_default_pod_node_selector(self, label_name: str, value: str): 
-    """Add a constraint for nodeSelector for a pipeline. Each constraint is a key-value pair label. For the 
+
+  def set_default_pod_node_selector(self, label_name: str, value: str):
+    """Add a constraint for nodeSelector for a pipeline. Each constraint is a key-value pair label. For the
       container to be eligible to run on a node, the node must have each of the constraints appeared
       as labels.
 
@@ -114,7 +114,7 @@
     """
     self.default_pod_node_selector[label_name] = value
     return self
-  
+
 
   def set_image_pull_policy(self, policy: str):
     """Configures the default image pull policy
@@ -128,9 +128,10 @@
 
   def add_op_transformer(self, transformer):
     """Configures the op_transformers which will be applied to all ops in the pipeline.
+    The ops can be ResourceOp, VolumenOp, or ContainerOp.
 
     Args:
-      transformer: a function that takes a ContainOp as input and returns a ContainerOp
+      transformer: a function that takes a kfp Op as input and returns a kfp Op
     """
     self.op_transformers.append(transformer)
 
diff --git a/sdk/python/kfp/onprem.py b/sdk/python/kfp/onprem.py
--- a/sdk/python/kfp/onprem.py
+++ b/sdk/python/kfp/onprem.py
@@ -1,7 +1,7 @@
 
 def mount_pvc(pvc_name='pipeline-claim', volume_name='pipeline', volume_mount_path='/mnt/pipeline'):
     """
-        Modifier function to apply to a Container Op to simplify volume, volume mount addition and 
+        Modifier function to apply to a Container Op to simplify volume, volume mount addition and
         enable better reuse of volumes, volume claims across container ops.
         Usage:
             train = train_op(...)
@@ -9,6 +9,10 @@
     """
     def _mount_pvc(task):
         from kubernetes import client as k8s_client
+        # there can be other ops in a pipeline (e.g. ResourceOp, VolumeOp)
+        # refer to #3906
+        if not hasattr(task, "add_volume") or not hasattr(task, "add_volume_mount"):
+            return task
         local_pvc = k8s_client.V1PersistentVolumeClaimVolumeSource(claim_name=pvc_name)
         return (
             task

{"golden_diff": "diff --git a/sdk/python/kfp/dsl/_pipeline.py b/sdk/python/kfp/dsl/_pipeline.py\n--- a/sdk/python/kfp/dsl/_pipeline.py\n+++ b/sdk/python/kfp/dsl/_pipeline.py\n@@ -102,9 +102,9 @@\n     \"\"\"\n     self.ttl_seconds_after_finished = seconds\n     return self\n-  \n-  def set_default_pod_node_selector(self, label_name: str, value: str): \n-    \"\"\"Add a constraint for nodeSelector for a pipeline. Each constraint is a key-value pair label. For the \n+\n+  def set_default_pod_node_selector(self, label_name: str, value: str):\n+    \"\"\"Add a constraint for nodeSelector for a pipeline. Each constraint is a key-value pair label. For the\n       container to be eligible to run on a node, the node must have each of the constraints appeared\n       as labels.\n \n@@ -114,7 +114,7 @@\n     \"\"\"\n     self.default_pod_node_selector[label_name] = value\n     return self\n-  \n+\n \n   def set_image_pull_policy(self, policy: str):\n     \"\"\"Configures the default image pull policy\n@@ -128,9 +128,10 @@\n \n   def add_op_transformer(self, transformer):\n     \"\"\"Configures the op_transformers which will be applied to all ops in the pipeline.\n+    The ops can be ResourceOp, VolumenOp, or ContainerOp.\n \n     Args:\n-      transformer: a function that takes a ContainOp as input and returns a ContainerOp\n+      transformer: a function that takes a kfp Op as input and returns a kfp Op\n     \"\"\"\n     self.op_transformers.append(transformer)\n \ndiff --git a/sdk/python/kfp/onprem.py b/sdk/python/kfp/onprem.py\n--- a/sdk/python/kfp/onprem.py\n+++ b/sdk/python/kfp/onprem.py\n@@ -1,7 +1,7 @@\n \n def mount_pvc(pvc_name='pipeline-claim', volume_name='pipeline', volume_mount_path='/mnt/pipeline'):\n     \"\"\"\n-        Modifier function to apply to a Container Op to simplify volume, volume mount addition and \n+        Modifier function to apply to a Container Op to simplify volume, volume mount addition and\n         enable better reuse of volumes, volume claims across container ops.\n         Usage:\n             train = train_op(...)\n@@ -9,6 +9,10 @@\n     \"\"\"\n     def _mount_pvc(task):\n         from kubernetes import client as k8s_client\n+        # there can be other ops in a pipeline (e.g. ResourceOp, VolumeOp)\n+        # refer to #3906\n+        if not hasattr(task, \"add_volume\") or not hasattr(task, \"add_volume_mount\"):\n+            return task\n         local_pvc = k8s_client.V1PersistentVolumeClaimVolumeSource(claim_name=pvc_name)\n         return (\n             task\n", "issue": "mount_pvc failing when pipeline includes ResourceOps\n### What steps did you take:\r\n\r\nCreated a persistent volume claim using a `VolumeOp`:\r\n\r\n```python\r\n    vop = dsl.VolumeOp(\r\n        name=\"shared-data\",\r\n        resource_name=\"shared-data\",\r\n        size=\"100G\",\r\n        storage_class=\"nfs\",\r\n    ).set_display_name(\"Prepare volume\")\r\n```\r\n\r\nAttempted to mount the PVC to every operation in a pipeline using:\r\n\r\n```python\r\n    dsl.get_pipeline_conf().add_op_transformer(\r\n        mount_pvc(pvc_name=vop.volume.persistent_volume_claim.claim_name, volume_mount_path=MOUNT_PATH)\r\n    )\r\n```\r\n\r\n### What happened:\r\nDuring compilation, the `add_op_transformer` attempts to invoke the `add_volume_mount` method on the `VolumeOp`, producing an error:\r\n\r\n```\r\nAttributeError: 'VolumeOp' object has no attribute 'add_volume_mount'\r\n```\r\n\r\n### What did you expect to happen:\r\n\r\nThe documentation states that transfomers take and return a `ContainerOp`:\r\n\r\nhttps://kubeflow-pipelines.readthedocs.io/en/latest/source/kfp.dsl.html#kfp.dsl.PipelineConf.add_op_transformer\r\n\r\nGiven that, I would expect them to be bypassed for `VolumeOp` (or other `ResourceOp` operations), and hence this compilation should succeed.\r\n\r\n### Environment:\r\n<!-- Please fill in those that seem relevant. -->\r\n\r\nKFP version: Build commit: 9c16e12\r\n\r\nKFP SDK version: <!-- Please attach the output of this shell command: $pip list | grep kfp -->\r\nkfp                      0.5.1\r\nkfp-server-api           0.3.0\r\n\r\n### Anything else you would like to add:\r\n[Miscellaneous information that will assist in solving the issue.]\r\n\r\n/kind bug\r\n<!-- Please include labels by uncommenting them to help us better triage issues, choose from the following -->\r\n<!--\r\n// /area frontend\r\n// /area backend\r\n// /area sdk\r\n// /area testing\r\n// /area engprod\r\n-->\r\n\n", "before_files": [{"content": "# Copyright 2018-2019 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\nfrom . import _container_op\nfrom . import _resource_op\nfrom . import _ops_group\nfrom ._component_bridge import _create_container_op_from_component_and_arguments\nfrom ..components import _components\nfrom ..components._naming import _make_name_unique_by_adding_index\nimport sys\n\n\n# This handler is called whenever the @pipeline decorator is applied.\n# It can be used by command-line DSL compiler to inject code that runs for every pipeline definition.\n_pipeline_decorator_handler = None\n\n\ndef pipeline(name : str = None, description : str = None):\n  \"\"\"Decorator of pipeline functions.\n\n  Usage:\n  ```python\n  @pipeline(\n    name='my awesome pipeline',\n    description='Is it really awesome?'\n  )\n  def my_pipeline(a: PipelineParam, b: PipelineParam):\n    ...\n  ```\n  \"\"\"\n  def _pipeline(func):\n    if name:\n      func._component_human_name = name\n    if description:\n      func._component_description = description\n\n    if _pipeline_decorator_handler:\n      return _pipeline_decorator_handler(func) or func\n    else:\n      return func\n\n  return _pipeline\n\nclass PipelineConf():\n  \"\"\"PipelineConf contains pipeline level settings\n  \"\"\"\n  def __init__(self):\n    self.image_pull_secrets = []\n    self.timeout = 0\n    self.ttl_seconds_after_finished = -1\n    self.op_transformers = []\n    self.default_pod_node_selector = {}\n    self.image_pull_policy = None\n    self.parallelism = None\n\n  def set_image_pull_secrets(self, image_pull_secrets):\n    \"\"\"Configures the pipeline level imagepullsecret\n\n    Args:\n      image_pull_secrets: a list of Kubernetes V1LocalObjectReference\n      For detailed description, check Kubernetes V1LocalObjectReference definition\n      https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1LocalObjectReference.md\n    \"\"\"\n    self.image_pull_secrets = image_pull_secrets\n    return self\n\n  def set_timeout(self, seconds: int):\n    \"\"\"Configures the pipeline level timeout\n\n    Args:\n      seconds: number of seconds for timeout\n    \"\"\"\n    self.timeout = seconds\n    return self\n\n  def set_parallelism(self, max_num_pods: int):\n    \"\"\"Configures the max number of total parallel pods that can execute at the same time in a workflow.\n\n    Args:\n        max_num_pods (int): max number of total parallel pods.\n    \"\"\"\n    self.parallelism = max_num_pods\n    return self\n\n  def set_ttl_seconds_after_finished(self, seconds: int):\n    \"\"\"Configures the ttl after the pipeline has finished.\n\n    Args:\n      seconds: number of seconds for the workflow to be garbage collected after it is finished.\n    \"\"\"\n    self.ttl_seconds_after_finished = seconds\n    return self\n  \n  def set_default_pod_node_selector(self, label_name: str, value: str): \n    \"\"\"Add a constraint for nodeSelector for a pipeline. Each constraint is a key-value pair label. For the \n      container to be eligible to run on a node, the node must have each of the constraints appeared\n      as labels.\n\n    Args:\n        label_name: The name of the constraint label.\n        value: The value of the constraint label.\n    \"\"\"\n    self.default_pod_node_selector[label_name] = value\n    return self\n  \n\n  def set_image_pull_policy(self, policy: str):\n    \"\"\"Configures the default image pull policy\n\n    Args:\n      policy: the pull policy, has to be one of: Always, Never, IfNotPresent.\n      For more info: https://github.com/kubernetes-client/python/blob/10a7f95435c0b94a6d949ba98375f8cc85a70e5a/kubernetes/docs/V1Container.md\n    \"\"\"\n    self.image_pull_policy = policy\n    return self\n\n  def add_op_transformer(self, transformer):\n    \"\"\"Configures the op_transformers which will be applied to all ops in the pipeline.\n\n    Args:\n      transformer: a function that takes a ContainOp as input and returns a ContainerOp\n    \"\"\"\n    self.op_transformers.append(transformer)\n\n\ndef get_pipeline_conf():\n  \"\"\"Configure the pipeline level setting to the current pipeline\n    Note: call the function inside the user defined pipeline function.\n  \"\"\"\n  return Pipeline.get_default_pipeline().conf\n\n#TODO: Pipeline is in fact an opsgroup, refactor the code.\nclass Pipeline():\n  \"\"\"A pipeline contains a list of operators.\n\n  This class is not supposed to be used by pipeline authors since pipeline authors can use\n  pipeline functions (decorated with @pipeline) to reference their pipelines. This class\n  is useful for implementing a compiler. For example, the compiler can use the following\n  to get the pipeline object and its ops:\n\n  ```python\n  with Pipeline() as p:\n    pipeline_func(*args_list)\n\n  traverse(p.ops)\n  ```\n  \"\"\"\n\n  # _default_pipeline is set when it (usually a compiler) runs \"with Pipeline()\"\n  _default_pipeline = None\n\n  @staticmethod\n  def get_default_pipeline():\n    \"\"\"Get default pipeline. \"\"\"\n    return Pipeline._default_pipeline\n\n  @staticmethod\n  def add_pipeline(name, description, func):\n    \"\"\"Add a pipeline function with the specified name and description.\"\"\"\n    # Applying the @pipeline decorator to the pipeline function\n    func = pipeline(name=name, description=description)(func)\n\n  def __init__(self, name: str):\n    \"\"\"Create a new instance of Pipeline.\n\n    Args:\n      name: the name of the pipeline. Once deployed, the name will show up in Pipeline System UI.\n    \"\"\"\n    self.name = name\n    self.ops = {}\n    # Add the root group.\n    self.groups = [_ops_group.OpsGroup('pipeline', name=name)]\n    self.group_id = 0\n    self.conf = PipelineConf()\n    self._metadata = None\n\n  def __enter__(self):\n    if Pipeline._default_pipeline:\n      raise Exception('Nested pipelines are not allowed.')\n\n    Pipeline._default_pipeline = self\n    self._old_container_task_constructor = _components._container_task_constructor\n    _components._container_task_constructor = _create_container_op_from_component_and_arguments\n\n    def register_op_and_generate_id(op):\n      return self.add_op(op, op.is_exit_handler)\n\n    self._old__register_op_handler = _container_op._register_op_handler\n    _container_op._register_op_handler = register_op_and_generate_id\n    return self\n\n  def __exit__(self, *args):\n    Pipeline._default_pipeline = None\n    _container_op._register_op_handler = self._old__register_op_handler\n    _components._container_task_constructor = self._old_container_task_constructor\n\n  def add_op(self, op: _container_op.BaseOp, define_only: bool):\n    \"\"\"Add a new operator.\n\n    Args:\n      op: An operator of ContainerOp, ResourceOp or their inherited types.\n\n    Returns\n      op_name: a unique op name.\n    \"\"\"\n    #If there is an existing op with this name then generate a new name.\n    op_name = _make_name_unique_by_adding_index(op.human_name, list(self.ops.keys()), ' ')\n\n    self.ops[op_name] = op\n    if not define_only:\n      self.groups[-1].ops.append(op)\n\n    return op_name\n\n  def push_ops_group(self, group: _ops_group.OpsGroup):\n    \"\"\"Push an OpsGroup into the stack.\n\n    Args:\n      group: An OpsGroup. Typically it is one of ExitHandler, Branch, and Loop.\n    \"\"\"\n    self.groups[-1].groups.append(group)\n    self.groups.append(group)\n\n  def pop_ops_group(self):\n    \"\"\"Remove the current OpsGroup from the stack.\"\"\"\n    del self.groups[-1]\n\n  def remove_op_from_groups(self, op):\n    for group in self.groups:\n      group.remove_op_recursive(op)\n\n  def get_next_group_id(self):\n    \"\"\"Get next id for a new group. \"\"\"\n\n    self.group_id += 1\n    return self.group_id\n\n  def _set_metadata(self, metadata):\n    '''_set_metadata passes the containerop the metadata information\n    Args:\n      metadata (ComponentMeta): component metadata\n    '''\n    self._metadata = metadata\n", "path": "sdk/python/kfp/dsl/_pipeline.py"}, {"content": "\ndef mount_pvc(pvc_name='pipeline-claim', volume_name='pipeline', volume_mount_path='/mnt/pipeline'):\n    \"\"\"\n        Modifier function to apply to a Container Op to simplify volume, volume mount addition and \n        enable better reuse of volumes, volume claims across container ops.\n        Usage:\n            train = train_op(...)\n            train.apply(mount_pvc('claim-name', 'pipeline', '/mnt/pipeline'))\n    \"\"\"\n    def _mount_pvc(task):\n        from kubernetes import client as k8s_client\n        local_pvc = k8s_client.V1PersistentVolumeClaimVolumeSource(claim_name=pvc_name)\n        return (\n            task\n                .add_volume(\n                    k8s_client.V1Volume(name=volume_name, persistent_volume_claim=local_pvc)\n                )\n                .add_volume_mount(\n                    k8s_client.V1VolumeMount(mount_path=volume_mount_path, name=volume_name)\n                )\n        )\n    return _mount_pvc\n", "path": "sdk/python/kfp/onprem.py"}], "after_files": [{"content": "# Copyright 2018-2019 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\nfrom . import _container_op\nfrom . import _resource_op\nfrom . import _ops_group\nfrom ._component_bridge import _create_container_op_from_component_and_arguments\nfrom ..components import _components\nfrom ..components._naming import _make_name_unique_by_adding_index\nimport sys\n\n\n# This handler is called whenever the @pipeline decorator is applied.\n# It can be used by command-line DSL compiler to inject code that runs for every pipeline definition.\n_pipeline_decorator_handler = None\n\n\ndef pipeline(name : str = None, description : str = None):\n  \"\"\"Decorator of pipeline functions.\n\n  Usage:\n  ```python\n  @pipeline(\n    name='my awesome pipeline',\n    description='Is it really awesome?'\n  )\n  def my_pipeline(a: PipelineParam, b: PipelineParam):\n    ...\n  ```\n  \"\"\"\n  def _pipeline(func):\n    if name:\n      func._component_human_name = name\n    if description:\n      func._component_description = description\n\n    if _pipeline_decorator_handler:\n      return _pipeline_decorator_handler(func) or func\n    else:\n      return func\n\n  return _pipeline\n\nclass PipelineConf():\n  \"\"\"PipelineConf contains pipeline level settings\n  \"\"\"\n  def __init__(self):\n    self.image_pull_secrets = []\n    self.timeout = 0\n    self.ttl_seconds_after_finished = -1\n    self.op_transformers = []\n    self.default_pod_node_selector = {}\n    self.image_pull_policy = None\n    self.parallelism = None\n\n  def set_image_pull_secrets(self, image_pull_secrets):\n    \"\"\"Configures the pipeline level imagepullsecret\n\n    Args:\n      image_pull_secrets: a list of Kubernetes V1LocalObjectReference\n      For detailed description, check Kubernetes V1LocalObjectReference definition\n      https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1LocalObjectReference.md\n    \"\"\"\n    self.image_pull_secrets = image_pull_secrets\n    return self\n\n  def set_timeout(self, seconds: int):\n    \"\"\"Configures the pipeline level timeout\n\n    Args:\n      seconds: number of seconds for timeout\n    \"\"\"\n    self.timeout = seconds\n    return self\n\n  def set_parallelism(self, max_num_pods: int):\n    \"\"\"Configures the max number of total parallel pods that can execute at the same time in a workflow.\n\n    Args:\n        max_num_pods (int): max number of total parallel pods.\n    \"\"\"\n    self.parallelism = max_num_pods\n    return self\n\n  def set_ttl_seconds_after_finished(self, seconds: int):\n    \"\"\"Configures the ttl after the pipeline has finished.\n\n    Args:\n      seconds: number of seconds for the workflow to be garbage collected after it is finished.\n    \"\"\"\n    self.ttl_seconds_after_finished = seconds\n    return self\n\n  def set_default_pod_node_selector(self, label_name: str, value: str):\n    \"\"\"Add a constraint for nodeSelector for a pipeline. Each constraint is a key-value pair label. For the\n      container to be eligible to run on a node, the node must have each of the constraints appeared\n      as labels.\n\n    Args:\n        label_name: The name of the constraint label.\n        value: The value of the constraint label.\n    \"\"\"\n    self.default_pod_node_selector[label_name] = value\n    return self\n\n\n  def set_image_pull_policy(self, policy: str):\n    \"\"\"Configures the default image pull policy\n\n    Args:\n      policy: the pull policy, has to be one of: Always, Never, IfNotPresent.\n      For more info: https://github.com/kubernetes-client/python/blob/10a7f95435c0b94a6d949ba98375f8cc85a70e5a/kubernetes/docs/V1Container.md\n    \"\"\"\n    self.image_pull_policy = policy\n    return self\n\n  def add_op_transformer(self, transformer):\n    \"\"\"Configures the op_transformers which will be applied to all ops in the pipeline.\n    The ops can be ResourceOp, VolumenOp, or ContainerOp.\n\n    Args:\n      transformer: a function that takes a kfp Op as input and returns a kfp Op\n    \"\"\"\n    self.op_transformers.append(transformer)\n\n\ndef get_pipeline_conf():\n  \"\"\"Configure the pipeline level setting to the current pipeline\n    Note: call the function inside the user defined pipeline function.\n  \"\"\"\n  return Pipeline.get_default_pipeline().conf\n\n#TODO: Pipeline is in fact an opsgroup, refactor the code.\nclass Pipeline():\n  \"\"\"A pipeline contains a list of operators.\n\n  This class is not supposed to be used by pipeline authors since pipeline authors can use\n  pipeline functions (decorated with @pipeline) to reference their pipelines. This class\n  is useful for implementing a compiler. For example, the compiler can use the following\n  to get the pipeline object and its ops:\n\n  ```python\n  with Pipeline() as p:\n    pipeline_func(*args_list)\n\n  traverse(p.ops)\n  ```\n  \"\"\"\n\n  # _default_pipeline is set when it (usually a compiler) runs \"with Pipeline()\"\n  _default_pipeline = None\n\n  @staticmethod\n  def get_default_pipeline():\n    \"\"\"Get default pipeline. \"\"\"\n    return Pipeline._default_pipeline\n\n  @staticmethod\n  def add_pipeline(name, description, func):\n    \"\"\"Add a pipeline function with the specified name and description.\"\"\"\n    # Applying the @pipeline decorator to the pipeline function\n    func = pipeline(name=name, description=description)(func)\n\n  def __init__(self, name: str):\n    \"\"\"Create a new instance of Pipeline.\n\n    Args:\n      name: the name of the pipeline. Once deployed, the name will show up in Pipeline System UI.\n    \"\"\"\n    self.name = name\n    self.ops = {}\n    # Add the root group.\n    self.groups = [_ops_group.OpsGroup('pipeline', name=name)]\n    self.group_id = 0\n    self.conf = PipelineConf()\n    self._metadata = None\n\n  def __enter__(self):\n    if Pipeline._default_pipeline:\n      raise Exception('Nested pipelines are not allowed.')\n\n    Pipeline._default_pipeline = self\n    self._old_container_task_constructor = _components._container_task_constructor\n    _components._container_task_constructor = _create_container_op_from_component_and_arguments\n\n    def register_op_and_generate_id(op):\n      return self.add_op(op, op.is_exit_handler)\n\n    self._old__register_op_handler = _container_op._register_op_handler\n    _container_op._register_op_handler = register_op_and_generate_id\n    return self\n\n  def __exit__(self, *args):\n    Pipeline._default_pipeline = None\n    _container_op._register_op_handler = self._old__register_op_handler\n    _components._container_task_constructor = self._old_container_task_constructor\n\n  def add_op(self, op: _container_op.BaseOp, define_only: bool):\n    \"\"\"Add a new operator.\n\n    Args:\n      op: An operator of ContainerOp, ResourceOp or their inherited types.\n\n    Returns\n      op_name: a unique op name.\n    \"\"\"\n    #If there is an existing op with this name then generate a new name.\n    op_name = _make_name_unique_by_adding_index(op.human_name, list(self.ops.keys()), ' ')\n\n    self.ops[op_name] = op\n    if not define_only:\n      self.groups[-1].ops.append(op)\n\n    return op_name\n\n  def push_ops_group(self, group: _ops_group.OpsGroup):\n    \"\"\"Push an OpsGroup into the stack.\n\n    Args:\n      group: An OpsGroup. Typically it is one of ExitHandler, Branch, and Loop.\n    \"\"\"\n    self.groups[-1].groups.append(group)\n    self.groups.append(group)\n\n  def pop_ops_group(self):\n    \"\"\"Remove the current OpsGroup from the stack.\"\"\"\n    del self.groups[-1]\n\n  def remove_op_from_groups(self, op):\n    for group in self.groups:\n      group.remove_op_recursive(op)\n\n  def get_next_group_id(self):\n    \"\"\"Get next id for a new group. \"\"\"\n\n    self.group_id += 1\n    return self.group_id\n\n  def _set_metadata(self, metadata):\n    '''_set_metadata passes the containerop the metadata information\n    Args:\n      metadata (ComponentMeta): component metadata\n    '''\n    self._metadata = metadata\n", "path": "sdk/python/kfp/dsl/_pipeline.py"}, {"content": "\ndef mount_pvc(pvc_name='pipeline-claim', volume_name='pipeline', volume_mount_path='/mnt/pipeline'):\n    \"\"\"\n        Modifier function to apply to a Container Op to simplify volume, volume mount addition and\n        enable better reuse of volumes, volume claims across container ops.\n        Usage:\n            train = train_op(...)\n            train.apply(mount_pvc('claim-name', 'pipeline', '/mnt/pipeline'))\n    \"\"\"\n    def _mount_pvc(task):\n        from kubernetes import client as k8s_client\n        # there can be other ops in a pipeline (e.g. ResourceOp, VolumeOp)\n        # refer to #3906\n        if not hasattr(task, \"add_volume\") or not hasattr(task, \"add_volume_mount\"):\n            return task\n        local_pvc = k8s_client.V1PersistentVolumeClaimVolumeSource(claim_name=pvc_name)\n        return (\n            task\n                .add_volume(\n                    k8s_client.V1Volume(name=volume_name, persistent_volume_claim=local_pvc)\n                )\n                .add_volume_mount(\n                    k8s_client.V1VolumeMount(mount_path=volume_mount_path, name=volume_name)\n                )\n        )\n    return _mount_pvc\n", "path": "sdk/python/kfp/onprem.py"}]}