We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
VolumeOp doesn't support GC after workflow deletion
The PVC created from VolumeOp is not automatically recycled after workflow is deleted.

I wonder if owner reference can work with PVC when it is set with workflow's reference. If it works, we can provide an option in volumeop to set it when creating the PVC.

If it doesn't work, we should provide a way to either use Argo's volume API or support a delete PVC op in exit handler.
VolumeOp doesn't support GC after workflow deletion
The PVC created from VolumeOp is not automatically recycled after workflow is deleted.

I wonder if owner reference can work with PVC when it is set with workflow's reference. If it works, we can provide an option in volumeop to set it when creating the PVC.

If it doesn't work, we should provide a way to either use Argo's volume API or support a delete PVC op in exit handler.

--- 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/_resource_op.py`
Content:
```
1 # Copyright 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 from typing import Dict, List, Optional
16 import warnings
17 
18 from ._container_op import BaseOp, ContainerOp
19 from . import _pipeline_param
20 
21 
22 class Resource(object):
23   """
24     A wrapper over Argo ResourceTemplate definition object
25     (io.argoproj.workflow.v1alpha1.ResourceTemplate)
26     which is used to represent the `resource` property in argo's workflow
27     template (io.argoproj.workflow.v1alpha1.Template).
28     """
29   swagger_types = {
30       "action": "str",
31       "merge_strategy": "str",
32       "success_condition": "str",
33       "failure_condition": "str",
34       "manifest": "str",
35       "flags": "list[str]"
36   }
37   openapi_types = {
38       "action": "str",
39       "merge_strategy": "str",
40       "success_condition": "str",
41       "failure_condition": "str",
42       "manifest": "str",
43       "flags": "list[str]"
44   }
45   attribute_map = {
46       "action": "action",
47       "merge_strategy": "mergeStrategy",
48       "success_condition": "successCondition",
49       "failure_condition": "failureCondition",
50       "manifest": "manifest",
51       "flags": "flags"
52   }
53 
54   def __init__(self,
55                action: str = None,
56                merge_strategy: str = None,
57                success_condition: str = None,
58                failure_condition: str = None,
59                manifest: str = None,
60                flags: Optional[List[str]] = None):
61     """Create a new instance of Resource"""
62     self.action = action
63     self.merge_strategy = merge_strategy
64     self.success_condition = success_condition
65     self.failure_condition = failure_condition
66     self.manifest = manifest
67     self.flags = flags
68 
69 
70 class ResourceOp(BaseOp):
71   """Represents an op which will be translated into a resource template.
72 
73     TODO(https://github.com/kubeflow/pipelines/issues/4822): Determine the
74         stability level of this feature.
75 
76     Args:
77       k8s_resource: A k8s resource which will be submitted to the cluster
78       action: One of "create"/"delete"/"apply"/"patch" (default is "create")
79       merge_strategy: The merge strategy for the "apply" action
80       success_condition: The successCondition of the template
81       failure_condition: The failureCondition of the template
82           For more info see:
83           https://github.com/argoproj/argo/blob/master/examples/k8s-jobs.yaml
84       attribute_outputs: Maps output labels to resource's json paths,
85         similarly to file_outputs of ContainerOp
86       kwargs: name, sidecars. See BaseOp definition
87 
88     Raises:
89       ValueError: if not inside a pipeline
90           if the name is an invalid string
91           if no k8s_resource is provided
92           if merge_strategy is set without "apply" action
93     """
94 
95   def __init__(self,
96                k8s_resource=None,
97                action: str = "create",
98                merge_strategy: str = None,
99                success_condition: str = None,
100                failure_condition: str = None,
101                attribute_outputs: Optional[Dict[str, str]] = None,
102                flags: Optional[List[str]] = None,
103                **kwargs):
104 
105     super().__init__(**kwargs)
106     self.attrs_with_pipelineparams = list(self.attrs_with_pipelineparams)
107     self.attrs_with_pipelineparams.extend(
108         ["_resource", "k8s_resource", "attribute_outputs"])
109 
110     if k8s_resource is None:
111       raise ValueError("You need to provide a k8s_resource.")
112 
113     if action == "delete":
114       warnings.warn(
115           'Please use `kubernetes_resource_delete_op` instead of '
116           '`ResourceOp(action="delete")`',
117           DeprecationWarning)
118 
119     if merge_strategy and action != "apply":
120       raise ValueError("You can't set merge_strategy when action != 'apply'")
121 
122     # if action is delete, there should not be any outputs, success_condition,
123     # and failure_condition
124     if action == "delete" and (success_condition or failure_condition or
125                                attribute_outputs):
126       raise ValueError(
127           "You can't set success_condition, failure_condition, or "
128           "attribute_outputs when action == 'delete'"
129       )
130 
131     if action == "delete" and flags is None:
132       flags = ["--wait=false"]
133     init_resource = {
134         "action": action,
135         "merge_strategy": merge_strategy,
136         "success_condition": success_condition,
137         "failure_condition": failure_condition,
138         "flags": flags
139     }
140     # `resource` prop in `io.argoproj.workflow.v1alpha1.Template`
141     self._resource = Resource(**init_resource)
142 
143     self.k8s_resource = k8s_resource
144 
145     # if action is delete, there should not be any outputs, success_condition,
146     # and failure_condition
147     if action == "delete":
148       self.attribute_outputs = {}
149       self.outputs = {}
150       self.output = None
151       return
152 
153     # Set attribute_outputs
154     extra_attribute_outputs = \
155         attribute_outputs if attribute_outputs else {}
156     self.attribute_outputs = \
157         self.attribute_outputs if hasattr(self, "attribute_outputs") \
158         else {}
159     self.attribute_outputs.update(extra_attribute_outputs)
160     # Add name and manifest if not specified by the user
161     if "name" not in self.attribute_outputs:
162       self.attribute_outputs["name"] = "{.metadata.name}"
163     if "manifest" not in self.attribute_outputs:
164       self.attribute_outputs["manifest"] = "{}"
165 
166     # Set outputs
167     self.outputs = {
168         name: _pipeline_param.PipelineParam(name, op_name=self.name)
169         for name in self.attribute_outputs.keys()
170     }
171     # If user set a single attribute_output, set self.output as that
172     # parameter, else set it as the resource name
173     self.output = self.outputs["name"]
174     if len(extra_attribute_outputs) == 1:
175       self.output = self.outputs[list(extra_attribute_outputs)[0]]
176 
177   @property
178   def resource(self):
179     """`Resource` object that represents the `resource` property in `io.argoproj.workflow.v1alpha1.Template`.
180     """
181     return self._resource
182 
183   def delete(self, flags: Optional[List[str]] = None):
184     """Returns a ResourceOp which deletes the resource."""
185     if self.resource.action == "delete":
186       raise ValueError("This operation is already a resource deletion.")
187 
188     if isinstance(self.k8s_resource, dict):
189       kind = self.k8s_resource["kind"]
190     else:
191       kind = self.k8s_resource.kind
192 
193     return kubernetes_resource_delete_op(
194         name=self.outputs["name"], kind=kind, flags=flags or ["--wait=false"])
195 
196 
197 def kubernetes_resource_delete_op(
198     name: str,
199     kind: str,
200     namespace: str = None,
201     flags: Optional[List[str]] = None,
202 ) -> ContainerOp:
203   """Operation that deletes a Kubernetes resource.
204 
205     Outputs:
206       name: The name of the deleted resource
207     """
208 
209   command = [
210       "kubectl", "delete",
211       str(kind),
212       str(name), "--ignore-not-found", "--output", "name"
213   ]
214   if namespace:
215     command.extend(["--namespace", str(namespace)])
216   if flags:
217     command.extend(flags)
218 
219   result = ContainerOp(
220       name="kubernetes_resource_delete",
221       image="gcr.io/cloud-builders/kubectl",
222       command=command,
223   )
224   return result
225
```
--- 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/_resource_op.py b/sdk/python/kfp/dsl/_resource_op.py
--- a/sdk/python/kfp/dsl/_resource_op.py
+++ b/sdk/python/kfp/dsl/_resource_op.py
@@ -31,6 +31,7 @@
       "merge_strategy": "str",
       "success_condition": "str",
       "failure_condition": "str",
+      "set_owner_reference": "bool",
       "manifest": "str",
       "flags": "list[str]"
   }
@@ -39,6 +40,7 @@
       "merge_strategy": "str",
       "success_condition": "str",
       "failure_condition": "str",
+      "set_owner_reference": "bool",
       "manifest": "str",
       "flags": "list[str]"
   }
@@ -47,6 +49,7 @@
       "merge_strategy": "mergeStrategy",
       "success_condition": "successCondition",
       "failure_condition": "failureCondition",
+      "set_owner_reference": "setOwnerReference",
       "manifest": "manifest",
       "flags": "flags"
   }
@@ -56,6 +59,7 @@
                merge_strategy: str = None,
                success_condition: str = None,
                failure_condition: str = None,
+               set_owner_reference: bool = None,
                manifest: str = None,
                flags: Optional[List[str]] = None):
     """Create a new instance of Resource"""
@@ -63,6 +67,7 @@
     self.merge_strategy = merge_strategy
     self.success_condition = success_condition
     self.failure_condition = failure_condition
+    self.set_owner_reference = set_owner_reference
     self.manifest = manifest
     self.flags = flags
 
@@ -98,6 +103,7 @@
                merge_strategy: str = None,
                success_condition: str = None,
                failure_condition: str = None,
+               set_owner_reference: bool = None,
                attribute_outputs: Optional[Dict[str, str]] = None,
                flags: Optional[List[str]] = None,
                **kwargs):
@@ -135,6 +141,7 @@
         "merge_strategy": merge_strategy,
         "success_condition": success_condition,
         "failure_condition": failure_condition,
+        "set_owner_reference": set_owner_reference,
         "flags": flags
     }
     # `resource` prop in `io.argoproj.workflow.v1alpha1.Template`

{"golden_diff": "diff --git a/sdk/python/kfp/dsl/_resource_op.py b/sdk/python/kfp/dsl/_resource_op.py\n--- a/sdk/python/kfp/dsl/_resource_op.py\n+++ b/sdk/python/kfp/dsl/_resource_op.py\n@@ -31,6 +31,7 @@\n       \"merge_strategy\": \"str\",\n       \"success_condition\": \"str\",\n       \"failure_condition\": \"str\",\n+      \"set_owner_reference\": \"bool\",\n       \"manifest\": \"str\",\n       \"flags\": \"list[str]\"\n   }\n@@ -39,6 +40,7 @@\n       \"merge_strategy\": \"str\",\n       \"success_condition\": \"str\",\n       \"failure_condition\": \"str\",\n+      \"set_owner_reference\": \"bool\",\n       \"manifest\": \"str\",\n       \"flags\": \"list[str]\"\n   }\n@@ -47,6 +49,7 @@\n       \"merge_strategy\": \"mergeStrategy\",\n       \"success_condition\": \"successCondition\",\n       \"failure_condition\": \"failureCondition\",\n+      \"set_owner_reference\": \"setOwnerReference\",\n       \"manifest\": \"manifest\",\n       \"flags\": \"flags\"\n   }\n@@ -56,6 +59,7 @@\n                merge_strategy: str = None,\n                success_condition: str = None,\n                failure_condition: str = None,\n+               set_owner_reference: bool = None,\n                manifest: str = None,\n                flags: Optional[List[str]] = None):\n     \"\"\"Create a new instance of Resource\"\"\"\n@@ -63,6 +67,7 @@\n     self.merge_strategy = merge_strategy\n     self.success_condition = success_condition\n     self.failure_condition = failure_condition\n+    self.set_owner_reference = set_owner_reference\n     self.manifest = manifest\n     self.flags = flags\n \n@@ -98,6 +103,7 @@\n                merge_strategy: str = None,\n                success_condition: str = None,\n                failure_condition: str = None,\n+               set_owner_reference: bool = None,\n                attribute_outputs: Optional[Dict[str, str]] = None,\n                flags: Optional[List[str]] = None,\n                **kwargs):\n@@ -135,6 +141,7 @@\n         \"merge_strategy\": merge_strategy,\n         \"success_condition\": success_condition,\n         \"failure_condition\": failure_condition,\n+        \"set_owner_reference\": set_owner_reference,\n         \"flags\": flags\n     }\n     # `resource` prop in `io.argoproj.workflow.v1alpha1.Template`\n", "issue": "VolumeOp doesn't support GC after workflow deletion\nThe PVC created from VolumeOp is not automatically recycled after workflow is deleted.\r\n\r\nI wonder if owner reference can work with PVC when it is set with workflow's reference. If it works, we can provide an option in volumeop to set it when creating the PVC.\r\n\r\nIf it doesn't work, we should provide a way to either use Argo's volume API or support a delete PVC op in exit handler.\nVolumeOp doesn't support GC after workflow deletion\nThe PVC created from VolumeOp is not automatically recycled after workflow is deleted.\r\n\r\nI wonder if owner reference can work with PVC when it is set with workflow's reference. If it works, we can provide an option in volumeop to set it when creating the PVC.\r\n\r\nIf it doesn't work, we should provide a way to either use Argo's volume API or support a delete PVC op in exit handler.\n", "before_files": [{"content": "# Copyright 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\nfrom typing import Dict, List, Optional\nimport warnings\n\nfrom ._container_op import BaseOp, ContainerOp\nfrom . import _pipeline_param\n\n\nclass Resource(object):\n  \"\"\"\n    A wrapper over Argo ResourceTemplate definition object\n    (io.argoproj.workflow.v1alpha1.ResourceTemplate)\n    which is used to represent the `resource` property in argo's workflow\n    template (io.argoproj.workflow.v1alpha1.Template).\n    \"\"\"\n  swagger_types = {\n      \"action\": \"str\",\n      \"merge_strategy\": \"str\",\n      \"success_condition\": \"str\",\n      \"failure_condition\": \"str\",\n      \"manifest\": \"str\",\n      \"flags\": \"list[str]\"\n  }\n  openapi_types = {\n      \"action\": \"str\",\n      \"merge_strategy\": \"str\",\n      \"success_condition\": \"str\",\n      \"failure_condition\": \"str\",\n      \"manifest\": \"str\",\n      \"flags\": \"list[str]\"\n  }\n  attribute_map = {\n      \"action\": \"action\",\n      \"merge_strategy\": \"mergeStrategy\",\n      \"success_condition\": \"successCondition\",\n      \"failure_condition\": \"failureCondition\",\n      \"manifest\": \"manifest\",\n      \"flags\": \"flags\"\n  }\n\n  def __init__(self,\n               action: str = None,\n               merge_strategy: str = None,\n               success_condition: str = None,\n               failure_condition: str = None,\n               manifest: str = None,\n               flags: Optional[List[str]] = None):\n    \"\"\"Create a new instance of Resource\"\"\"\n    self.action = action\n    self.merge_strategy = merge_strategy\n    self.success_condition = success_condition\n    self.failure_condition = failure_condition\n    self.manifest = manifest\n    self.flags = flags\n\n\nclass ResourceOp(BaseOp):\n  \"\"\"Represents an op which will be translated into a resource template.\n\n    TODO(https://github.com/kubeflow/pipelines/issues/4822): Determine the\n        stability level of this feature.\n\n    Args:\n      k8s_resource: A k8s resource which will be submitted to the cluster\n      action: One of \"create\"/\"delete\"/\"apply\"/\"patch\" (default is \"create\")\n      merge_strategy: The merge strategy for the \"apply\" action\n      success_condition: The successCondition of the template\n      failure_condition: The failureCondition of the template\n          For more info see:\n          https://github.com/argoproj/argo/blob/master/examples/k8s-jobs.yaml\n      attribute_outputs: Maps output labels to resource's json paths,\n        similarly to file_outputs of ContainerOp\n      kwargs: name, sidecars. See BaseOp definition\n\n    Raises:\n      ValueError: if not inside a pipeline\n          if the name is an invalid string\n          if no k8s_resource is provided\n          if merge_strategy is set without \"apply\" action\n    \"\"\"\n\n  def __init__(self,\n               k8s_resource=None,\n               action: str = \"create\",\n               merge_strategy: str = None,\n               success_condition: str = None,\n               failure_condition: str = None,\n               attribute_outputs: Optional[Dict[str, str]] = None,\n               flags: Optional[List[str]] = None,\n               **kwargs):\n\n    super().__init__(**kwargs)\n    self.attrs_with_pipelineparams = list(self.attrs_with_pipelineparams)\n    self.attrs_with_pipelineparams.extend(\n        [\"_resource\", \"k8s_resource\", \"attribute_outputs\"])\n\n    if k8s_resource is None:\n      raise ValueError(\"You need to provide a k8s_resource.\")\n\n    if action == \"delete\":\n      warnings.warn(\n          'Please use `kubernetes_resource_delete_op` instead of '\n          '`ResourceOp(action=\"delete\")`',\n          DeprecationWarning)\n\n    if merge_strategy and action != \"apply\":\n      raise ValueError(\"You can't set merge_strategy when action != 'apply'\")\n\n    # if action is delete, there should not be any outputs, success_condition,\n    # and failure_condition\n    if action == \"delete\" and (success_condition or failure_condition or\n                               attribute_outputs):\n      raise ValueError(\n          \"You can't set success_condition, failure_condition, or \"\n          \"attribute_outputs when action == 'delete'\"\n      )\n\n    if action == \"delete\" and flags is None:\n      flags = [\"--wait=false\"]\n    init_resource = {\n        \"action\": action,\n        \"merge_strategy\": merge_strategy,\n        \"success_condition\": success_condition,\n        \"failure_condition\": failure_condition,\n        \"flags\": flags\n    }\n    # `resource` prop in `io.argoproj.workflow.v1alpha1.Template`\n    self._resource = Resource(**init_resource)\n\n    self.k8s_resource = k8s_resource\n\n    # if action is delete, there should not be any outputs, success_condition,\n    # and failure_condition\n    if action == \"delete\":\n      self.attribute_outputs = {}\n      self.outputs = {}\n      self.output = None\n      return\n\n    # Set attribute_outputs\n    extra_attribute_outputs = \\\n        attribute_outputs if attribute_outputs else {}\n    self.attribute_outputs = \\\n        self.attribute_outputs if hasattr(self, \"attribute_outputs\") \\\n        else {}\n    self.attribute_outputs.update(extra_attribute_outputs)\n    # Add name and manifest if not specified by the user\n    if \"name\" not in self.attribute_outputs:\n      self.attribute_outputs[\"name\"] = \"{.metadata.name}\"\n    if \"manifest\" not in self.attribute_outputs:\n      self.attribute_outputs[\"manifest\"] = \"{}\"\n\n    # Set outputs\n    self.outputs = {\n        name: _pipeline_param.PipelineParam(name, op_name=self.name)\n        for name in self.attribute_outputs.keys()\n    }\n    # If user set a single attribute_output, set self.output as that\n    # parameter, else set it as the resource name\n    self.output = self.outputs[\"name\"]\n    if len(extra_attribute_outputs) == 1:\n      self.output = self.outputs[list(extra_attribute_outputs)[0]]\n\n  @property\n  def resource(self):\n    \"\"\"`Resource` object that represents the `resource` property in `io.argoproj.workflow.v1alpha1.Template`.\n    \"\"\"\n    return self._resource\n\n  def delete(self, flags: Optional[List[str]] = None):\n    \"\"\"Returns a ResourceOp which deletes the resource.\"\"\"\n    if self.resource.action == \"delete\":\n      raise ValueError(\"This operation is already a resource deletion.\")\n\n    if isinstance(self.k8s_resource, dict):\n      kind = self.k8s_resource[\"kind\"]\n    else:\n      kind = self.k8s_resource.kind\n\n    return kubernetes_resource_delete_op(\n        name=self.outputs[\"name\"], kind=kind, flags=flags or [\"--wait=false\"])\n\n\ndef kubernetes_resource_delete_op(\n    name: str,\n    kind: str,\n    namespace: str = None,\n    flags: Optional[List[str]] = None,\n) -> ContainerOp:\n  \"\"\"Operation that deletes a Kubernetes resource.\n\n    Outputs:\n      name: The name of the deleted resource\n    \"\"\"\n\n  command = [\n      \"kubectl\", \"delete\",\n      str(kind),\n      str(name), \"--ignore-not-found\", \"--output\", \"name\"\n  ]\n  if namespace:\n    command.extend([\"--namespace\", str(namespace)])\n  if flags:\n    command.extend(flags)\n\n  result = ContainerOp(\n      name=\"kubernetes_resource_delete\",\n      image=\"gcr.io/cloud-builders/kubectl\",\n      command=command,\n  )\n  return result\n", "path": "sdk/python/kfp/dsl/_resource_op.py"}], "after_files": [{"content": "# Copyright 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\nfrom typing import Dict, List, Optional\nimport warnings\n\nfrom ._container_op import BaseOp, ContainerOp\nfrom . import _pipeline_param\n\n\nclass Resource(object):\n  \"\"\"\n    A wrapper over Argo ResourceTemplate definition object\n    (io.argoproj.workflow.v1alpha1.ResourceTemplate)\n    which is used to represent the `resource` property in argo's workflow\n    template (io.argoproj.workflow.v1alpha1.Template).\n    \"\"\"\n  swagger_types = {\n      \"action\": \"str\",\n      \"merge_strategy\": \"str\",\n      \"success_condition\": \"str\",\n      \"failure_condition\": \"str\",\n      \"set_owner_reference\": \"bool\",\n      \"manifest\": \"str\",\n      \"flags\": \"list[str]\"\n  }\n  openapi_types = {\n      \"action\": \"str\",\n      \"merge_strategy\": \"str\",\n      \"success_condition\": \"str\",\n      \"failure_condition\": \"str\",\n      \"set_owner_reference\": \"bool\",\n      \"manifest\": \"str\",\n      \"flags\": \"list[str]\"\n  }\n  attribute_map = {\n      \"action\": \"action\",\n      \"merge_strategy\": \"mergeStrategy\",\n      \"success_condition\": \"successCondition\",\n      \"failure_condition\": \"failureCondition\",\n      \"set_owner_reference\": \"setOwnerReference\",\n      \"manifest\": \"manifest\",\n      \"flags\": \"flags\"\n  }\n\n  def __init__(self,\n               action: str = None,\n               merge_strategy: str = None,\n               success_condition: str = None,\n               failure_condition: str = None,\n               set_owner_reference: bool = None,\n               manifest: str = None,\n               flags: Optional[List[str]] = None):\n    \"\"\"Create a new instance of Resource\"\"\"\n    self.action = action\n    self.merge_strategy = merge_strategy\n    self.success_condition = success_condition\n    self.failure_condition = failure_condition\n    self.set_owner_reference = set_owner_reference\n    self.manifest = manifest\n    self.flags = flags\n\n\nclass ResourceOp(BaseOp):\n  \"\"\"Represents an op which will be translated into a resource template.\n\n    TODO(https://github.com/kubeflow/pipelines/issues/4822): Determine the\n        stability level of this feature.\n\n    Args:\n      k8s_resource: A k8s resource which will be submitted to the cluster\n      action: One of \"create\"/\"delete\"/\"apply\"/\"patch\" (default is \"create\")\n      merge_strategy: The merge strategy for the \"apply\" action\n      success_condition: The successCondition of the template\n      failure_condition: The failureCondition of the template\n          For more info see:\n          https://github.com/argoproj/argo/blob/master/examples/k8s-jobs.yaml\n      attribute_outputs: Maps output labels to resource's json paths,\n        similarly to file_outputs of ContainerOp\n      kwargs: name, sidecars. See BaseOp definition\n\n    Raises:\n      ValueError: if not inside a pipeline\n          if the name is an invalid string\n          if no k8s_resource is provided\n          if merge_strategy is set without \"apply\" action\n    \"\"\"\n\n  def __init__(self,\n               k8s_resource=None,\n               action: str = \"create\",\n               merge_strategy: str = None,\n               success_condition: str = None,\n               failure_condition: str = None,\n               set_owner_reference: bool = None,\n               attribute_outputs: Optional[Dict[str, str]] = None,\n               flags: Optional[List[str]] = None,\n               **kwargs):\n\n    super().__init__(**kwargs)\n    self.attrs_with_pipelineparams = list(self.attrs_with_pipelineparams)\n    self.attrs_with_pipelineparams.extend(\n        [\"_resource\", \"k8s_resource\", \"attribute_outputs\"])\n\n    if k8s_resource is None:\n      raise ValueError(\"You need to provide a k8s_resource.\")\n\n    if action == \"delete\":\n      warnings.warn(\n          'Please use `kubernetes_resource_delete_op` instead of '\n          '`ResourceOp(action=\"delete\")`',\n          DeprecationWarning)\n\n    if merge_strategy and action != \"apply\":\n      raise ValueError(\"You can't set merge_strategy when action != 'apply'\")\n\n    # if action is delete, there should not be any outputs, success_condition,\n    # and failure_condition\n    if action == \"delete\" and (success_condition or failure_condition or\n                               attribute_outputs):\n      raise ValueError(\n          \"You can't set success_condition, failure_condition, or \"\n          \"attribute_outputs when action == 'delete'\"\n      )\n\n    if action == \"delete\" and flags is None:\n      flags = [\"--wait=false\"]\n    init_resource = {\n        \"action\": action,\n        \"merge_strategy\": merge_strategy,\n        \"success_condition\": success_condition,\n        \"failure_condition\": failure_condition,\n        \"set_owner_reference\": set_owner_reference,\n        \"flags\": flags\n    }\n    # `resource` prop in `io.argoproj.workflow.v1alpha1.Template`\n    self._resource = Resource(**init_resource)\n\n    self.k8s_resource = k8s_resource\n\n    # if action is delete, there should not be any outputs, success_condition,\n    # and failure_condition\n    if action == \"delete\":\n      self.attribute_outputs = {}\n      self.outputs = {}\n      self.output = None\n      return\n\n    # Set attribute_outputs\n    extra_attribute_outputs = \\\n        attribute_outputs if attribute_outputs else {}\n    self.attribute_outputs = \\\n        self.attribute_outputs if hasattr(self, \"attribute_outputs\") \\\n        else {}\n    self.attribute_outputs.update(extra_attribute_outputs)\n    # Add name and manifest if not specified by the user\n    if \"name\" not in self.attribute_outputs:\n      self.attribute_outputs[\"name\"] = \"{.metadata.name}\"\n    if \"manifest\" not in self.attribute_outputs:\n      self.attribute_outputs[\"manifest\"] = \"{}\"\n\n    # Set outputs\n    self.outputs = {\n        name: _pipeline_param.PipelineParam(name, op_name=self.name)\n        for name in self.attribute_outputs.keys()\n    }\n    # If user set a single attribute_output, set self.output as that\n    # parameter, else set it as the resource name\n    self.output = self.outputs[\"name\"]\n    if len(extra_attribute_outputs) == 1:\n      self.output = self.outputs[list(extra_attribute_outputs)[0]]\n\n  @property\n  def resource(self):\n    \"\"\"`Resource` object that represents the `resource` property in `io.argoproj.workflow.v1alpha1.Template`.\n    \"\"\"\n    return self._resource\n\n  def delete(self, flags: Optional[List[str]] = None):\n    \"\"\"Returns a ResourceOp which deletes the resource.\"\"\"\n    if self.resource.action == \"delete\":\n      raise ValueError(\"This operation is already a resource deletion.\")\n\n    if isinstance(self.k8s_resource, dict):\n      kind = self.k8s_resource[\"kind\"]\n    else:\n      kind = self.k8s_resource.kind\n\n    return kubernetes_resource_delete_op(\n        name=self.outputs[\"name\"], kind=kind, flags=flags or [\"--wait=false\"])\n\n\ndef kubernetes_resource_delete_op(\n    name: str,\n    kind: str,\n    namespace: str = None,\n    flags: Optional[List[str]] = None,\n) -> ContainerOp:\n  \"\"\"Operation that deletes a Kubernetes resource.\n\n    Outputs:\n      name: The name of the deleted resource\n    \"\"\"\n\n  command = [\n      \"kubectl\", \"delete\",\n      str(kind),\n      str(name), \"--ignore-not-found\", \"--output\", \"name\"\n  ]\n  if namespace:\n    command.extend([\"--namespace\", str(namespace)])\n  if flags:\n    command.extend(flags)\n\n  result = ContainerOp(\n      name=\"kubernetes_resource_delete\",\n      image=\"gcr.io/cloud-builders/kubectl\",\n      command=command,\n  )\n  return result\n", "path": "sdk/python/kfp/dsl/_resource_op.py"}]}