From f5c4a6ff6a9e4942bd499ae18568dfa463a4e251 Mon Sep 17 00:00:00 2001
From: raphaelauv <raphaelauv@gmail.com>
Date: Sun, 5 Mar 2023 00:01:20 +0100
Subject: [PATCH 01/10] KubernetesResourceOperator - create PVC & delete PVC

---
 .../cncf/kubernetes/operators/resource.py     | 108 ++++++++++++++++++
 1 file changed, 108 insertions(+)
 create mode 100644 airflow/providers/cncf/kubernetes/operators/resource.py

diff --git a/airflow/providers/cncf/kubernetes/operators/resource.py b/airflow/providers/cncf/kubernetes/operators/resource.py
new file mode 100644
index 0000000000000..057c28ee3f0ac
--- /dev/null
+++ b/airflow/providers/cncf/kubernetes/operators/resource.py
@@ -0,0 +1,108 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Manage a Kubernetes Resource"""
+
+from __future__ import annotations
+
+from typing import Sequence
+
+from kubernetes.client import CoreV1Api
+
+from airflow.compat.functools import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.utils import yaml
+
+
+class KubernetesResourceOperator(BaseOperator):
+    """Abstract base class for all Kubernetes Resource operators."""
+
+    template_fields: Sequence[str] = ("resource_name",)
+    template_fields_renderers = {}
+
+    def __init__(
+        self,
+        *,
+        namespace: str | None = None,
+        resource_name: str | None = None,
+        kubernetes_conn_id: str | None = "kubernetes_default",
+        in_cluster: bool | None = None,
+        cluster_context: str | None = None,
+        config_file: str | None = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self._namespace = namespace
+        self.resource_name = resource_name
+        self.kubernetes_conn_id = kubernetes_conn_id
+        self.in_cluster = in_cluster
+        self.cluster_context = cluster_context
+        self.config_file = config_file
+
+    @cached_property
+    def client(self) -> CoreV1Api:
+        return self.hook.core_v1_client
+
+    @cached_property
+    def hook(self) -> KubernetesHook:
+        hook = KubernetesHook(
+            conn_id=self.kubernetes_conn_id,
+            in_cluster=self.in_cluster,
+            config_file=self.config_file,
+            cluster_context=self.cluster_context,
+        )
+        return hook
+
+    def get_namespace(self) -> str:
+        if self._namespace:
+            return self._namespace
+        else:
+            tmp = self.hook.get_namespace()
+            if tmp:
+                return tmp
+            else:
+                return "default"
+
+
+class kubernetesPVCcreate(KubernetesResourceOperator):
+    """Create PVC resource in a kubernetes."""
+
+    template_fields: Sequence[str] = tuple(
+        {
+            "pvc_conf",
+        }
+        | set(KubernetesResourceOperator.template_fields)
+    )
+    pvc_conf: str
+
+    def __int__(self, *, pvc_conf: str, **kwargs):
+        super().__init__(**kwargs)
+        self.pvc_conf = pvc_conf
+
+    def execute(self, context) -> None:
+        self.client.create_namespaced_persistent_volume_claim(
+            namespace=self.get_namespace(), body=yaml.safe_load(self.pvc_conf)
+        )
+
+
+class kubernetesPVCdelete(KubernetesResourceOperator):
+    """Delete PVC resource in a kubernetes."""
+
+    def execute(self, context) -> None:
+        self.client.delete_namespaced_persistent_volume_claim(
+            name=self.resource_name, namespace=self.get_namespace()
+        )

From 0a1dd12082444282c68935a949a981afa67bed69 Mon Sep 17 00:00:00 2001
From: raphaelauv <raphaelauv@gmail.com>
Date: Sun, 5 Mar 2023 15:10:19 +0100
Subject: [PATCH 02/10] fix

---
 .../cncf/kubernetes/operators/resource.py     |  54 +++----
 .../providers/cncf/kubernetes/provider.yaml   |   1 +
 .../cncf/kubernetes/utils/delete_from.py      | 141 ++++++++++++++++++
 tests/always/test_project_structure.py        |   5 +
 .../kubernetes/operators/test_resource.py     | 122 +++++++++++++++
 .../kubernetes/example_kubernetes_resource.py |  97 ++++++++++++
 6 files changed, 390 insertions(+), 30 deletions(-)
 create mode 100644 airflow/providers/cncf/kubernetes/utils/delete_from.py
 create mode 100644 tests/providers/cncf/kubernetes/operators/test_resource.py
 create mode 100644 tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py

diff --git a/airflow/providers/cncf/kubernetes/operators/resource.py b/airflow/providers/cncf/kubernetes/operators/resource.py
index 057c28ee3f0ac..654d1860d827d 100644
--- a/airflow/providers/cncf/kubernetes/operators/resource.py
+++ b/airflow/providers/cncf/kubernetes/operators/resource.py
@@ -18,27 +18,27 @@
 
 from __future__ import annotations
 
-from typing import Sequence
-
-from kubernetes.client import CoreV1Api
+from kubernetes.client import ApiClient
+from kubernetes.utils import create_from_yaml
 
 from airflow.compat.functools import cached_property
 from airflow.models import BaseOperator
 from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.utils.delete_from import delete_from_dict
 from airflow.utils import yaml
 
 
-class KubernetesResourceOperator(BaseOperator):
+class KubernetesResourceBaseOperator(BaseOperator):
     """Abstract base class for all Kubernetes Resource operators."""
 
-    template_fields: Sequence[str] = ("resource_name",)
-    template_fields_renderers = {}
+    template_fields = ("yaml_conf",)
+    template_fields_renderers = {"yaml_conf": "yaml"}
 
     def __init__(
         self,
         *,
+        yaml_conf: str,
         namespace: str | None = None,
-        resource_name: str | None = None,
         kubernetes_conn_id: str | None = "kubernetes_default",
         in_cluster: bool | None = None,
         cluster_context: str | None = None,
@@ -47,15 +47,15 @@ def __init__(
     ) -> None:
         super().__init__(**kwargs)
         self._namespace = namespace
-        self.resource_name = resource_name
         self.kubernetes_conn_id = kubernetes_conn_id
         self.in_cluster = in_cluster
         self.cluster_context = cluster_context
         self.config_file = config_file
+        self.yaml_conf = yaml_conf
 
     @cached_property
-    def client(self) -> CoreV1Api:
-        return self.hook.core_v1_client
+    def client(self) -> ApiClient:
+        return self.hook.api_client
 
     @cached_property
     def hook(self) -> KubernetesHook:
@@ -78,31 +78,25 @@ def get_namespace(self) -> str:
                 return "default"
 
 
-class kubernetesPVCcreate(KubernetesResourceOperator):
-    """Create PVC resource in a kubernetes."""
-
-    template_fields: Sequence[str] = tuple(
-        {
-            "pvc_conf",
-        }
-        | set(KubernetesResourceOperator.template_fields)
-    )
-    pvc_conf: str
-
-    def __int__(self, *, pvc_conf: str, **kwargs):
-        super().__init__(**kwargs)
-        self.pvc_conf = pvc_conf
+class KubernetesCreateResourceOperator(KubernetesResourceBaseOperator):
+    """Create a resource in a kubernetes."""
 
     def execute(self, context) -> None:
-        self.client.create_namespaced_persistent_volume_claim(
-            namespace=self.get_namespace(), body=yaml.safe_load(self.pvc_conf)
+        print(self.yaml_conf)
+        create_from_yaml(
+            k8s_client=self.client,
+            yaml_objects=[yaml.safe_load(self.yaml_conf)],
+            namespace=self.get_namespace(),
         )
 
 
-class kubernetesPVCdelete(KubernetesResourceOperator):
-    """Delete PVC resource in a kubernetes."""
+class KubernetesDeleteResourceOperator(KubernetesResourceBaseOperator):
+    """Delete a resource in a kubernetes."""
 
     def execute(self, context) -> None:
-        self.client.delete_namespaced_persistent_volume_claim(
-            name=self.resource_name, namespace=self.get_namespace()
+        print(self.yaml_conf)
+        delete_from_dict(
+            k8s_client=self.client,
+            yml_document=yaml.safe_load(self.yaml_conf),
+            namespace=self.get_namespace(),
         )
diff --git a/airflow/providers/cncf/kubernetes/provider.yaml b/airflow/providers/cncf/kubernetes/provider.yaml
index 34aa2f137f666..b43b3b223b9dd 100644
--- a/airflow/providers/cncf/kubernetes/provider.yaml
+++ b/airflow/providers/cncf/kubernetes/provider.yaml
@@ -96,6 +96,7 @@ operators:
       - airflow.providers.cncf.kubernetes.operators.kubernetes_pod
       - airflow.providers.cncf.kubernetes.operators.pod
       - airflow.providers.cncf.kubernetes.operators.spark_kubernetes
+      - airflow.providers.cncf.kubernetes.operators.resource
 
 sensors:
   - integration-name: Kubernetes
diff --git a/airflow/providers/cncf/kubernetes/utils/delete_from.py b/airflow/providers/cncf/kubernetes/utils/delete_from.py
new file mode 100644
index 0000000000000..eb2d235a6f956
--- /dev/null
+++ b/airflow/providers/cncf/kubernetes/utils/delete_from.py
@@ -0,0 +1,141 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# from https://github.com/tomplus/kubernetes_asyncio/pull/239/files
+
+from __future__ import annotations
+
+import re
+
+from kubernetes import client
+
+DEFAULT_DELETION_BODY = client.V1DeleteOptions(
+    propagation_policy="Background",
+    grace_period_seconds=5,
+)
+
+
+def delete_from_dict(
+    k8s_client,
+    yml_document,
+    verbose=False,
+    namespace="default",
+    body=None,
+    **kwargs,
+):
+
+    if body is None:
+        body = DEFAULT_DELETION_BODY
+
+    api_exceptions = []
+    if "List" in yml_document["kind"]:
+        kind = yml_document["kind"].replace("List", "")
+        for yml_doc in yml_document["items"]:
+            if kind != "":
+                yml_doc["apiVersion"] = yml_document["apiVersion"]
+                yml_doc["kind"] = kind
+            try:
+                _delete_from_yaml_single_item(
+                    k8s_client,
+                    yml_doc,
+                    verbose,
+                    namespace=namespace,
+                    body=body,
+                    **kwargs,
+                )
+            except client.rest.ApiException as api_exception:
+                api_exceptions.append(api_exception)
+    else:
+
+        try:
+            _delete_from_yaml_single_item(
+                k8s_client,
+                yml_document,
+                verbose,
+                namespace=namespace,
+                body=body,
+                **kwargs,
+            )
+        except client.rest.ApiException as api_exception:
+            api_exceptions.append(api_exception)
+
+    if api_exceptions:
+        raise FailToDeleteError(api_exceptions)
+
+
+def _delete_from_yaml_single_item(
+    k8s_client,
+    yml_document,
+    verbose=False,
+    namespace="default",
+    body=None,
+    **kwargs,
+):
+    if body is None:
+        body = DEFAULT_DELETION_BODY
+
+    # get group and version from apiVersion
+    group, _, version = yml_document["apiVersion"].partition("/")
+    if version == "":
+        version = group
+        group = "core"
+    # Take care for the case e.g. api_type is "apiextensions.k8s.io"
+    # Only replace the last instance
+    group = "".join(group.rsplit(".k8s.io", 1))
+    # convert group name from DNS subdomain format to
+    # python class name convention
+    group = "".join(word.capitalize() for word in group.split("."))
+    fcn_to_call = f"{group}{version.capitalize()}Api"
+    k8s_api = getattr(client, fcn_to_call)(k8s_client)
+    # Replace CamelCased action_type into snake_case
+    kind = yml_document["kind"]
+    kind = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", kind)
+    kind = re.sub("([a-z0-9])([A-Z])", r"\1_\2", kind).lower()
+
+    # Decide which namespace we are going to use for deleting the object
+    # IMPORTANT: Its ignore namespace in args:
+    #    create_from_yaml_single_item have same behaviour
+    if "namespace" in yml_document["metadata"]:
+        namespace = yml_document["metadata"]["namespace"]
+    name = yml_document["metadata"]["name"]
+
+    # Expect the user to delete namespaced objects more often
+    if hasattr(k8s_api, f"delete_namespaced_{kind}"):
+        resp: client.V1Status = getattr(k8s_api, f"delete_namespaced_{kind}")(
+            name=name, namespace=namespace, body=body, **kwargs
+        )
+    else:
+        resp: client.V1Status = getattr(k8s_api, f"delete_{kind}")(name=name, body=body, **kwargs)
+    if verbose:
+        print(f"{kind} deleted. status='{str(resp.status)}'")
+    return resp
+
+
+class FailToDeleteError(Exception):
+    """
+    An exception class for handling error if an error occurred when
+    handling a yaml file during deletion of the resource.
+    """
+
+    def __init__(self, api_exceptions):
+        self.api_exceptions = api_exceptions
+
+    def __str__(self):
+        msg = ""
+        for api_exception in self.api_exceptions:
+            msg += f"Error from server ({api_exception.reason}):{api_exception.body}"
+        return msg
diff --git a/tests/always/test_project_structure.py b/tests/always/test_project_structure.py
index 9f0d36b4e2d61..75444994876a7 100644
--- a/tests/always/test_project_structure.py
+++ b/tests/always/test_project_structure.py
@@ -428,6 +428,11 @@ class TestCncfProviderProjectStructure(ExampleCoverageTest):
         "airflow.providers.cncf.kubernetes.operators.kubernetes_pod",
         "airflow.providers.cncf.kubernetes.triggers.kubernetes_pod",
     }
+    BASE_CLASSES = {"airflow.providers.cncf.kubernetes.operators.resource.KubernetesResourceBaseOperator"}
+
+    MISSING_EXAMPLES_FOR_CLASSES = {
+        "airflow.providers.cncf.kubernetes.operators.resource.KubernetesResourceBaseOperator"
+    }
 
 
 class TestSlackProviderProjectStructure(ExampleCoverageTest):
diff --git a/tests/providers/cncf/kubernetes/operators/test_resource.py b/tests/providers/cncf/kubernetes/operators/test_resource.py
new file mode 100644
index 0000000000000..5d3b7423d6fcd
--- /dev/null
+++ b/tests/providers/cncf/kubernetes/operators/test_resource.py
@@ -0,0 +1,122 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from unittest.mock import patch
+
+import yaml
+
+from airflow import DAG
+from airflow.models import Connection
+from airflow.providers.cncf.kubernetes.operators.resource import (
+    KubernetesCreateResourceOperator,
+    KubernetesDeleteResourceOperator,
+)
+from airflow.utils import db, timezone
+
+TEST_VALID_RESOURCE_YAML = """
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: test_pvc
+spec:
+  accessModes:
+    - ReadWriteOnce
+  storageClassName: standard-rwo
+  resources:
+    requests:
+      storage: 500Gi
+"""
+
+
+@patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.KubernetesHook.get_conn")
+@patch("kubernetes.client.api.CoreV1Api.create_namespaced_persistent_volume_claim")
+@patch("kubernetes.client.api.CoreV1Api.delete_namespaced_persistent_volume_claim")
+@patch("airflow.utils.context.Context")
+class TestKubernetesCreateOperator:
+    def setup_method(self):
+        db.merge_conn(
+            Connection(
+                conn_id="kubernetes_default_kube_config",
+                conn_type="kubernetes",
+                extra=json.dumps({}),
+            )
+        )
+
+        db.merge_conn(
+            Connection(
+                conn_id="kubernetes_with_namespace",
+                conn_type="kubernetes",
+                extra=json.dumps({"namespace": "mock_namespace"}),
+            )
+        )
+
+        args = {"owner": "airflow", "start_date": timezone.datetime(2020, 2, 1)}
+        self.dag = DAG("test_dag_id", default_args=args)
+
+    def test_create_application_from_yaml(
+        self,
+        context,
+        mock_delete_namespaced_persistent_volume_claim,
+        mock_create_namespaced_persistent_volume_claim,
+        mock_kubernetes_hook,
+    ):
+        op = KubernetesCreateResourceOperator(
+            yaml_conf=TEST_VALID_RESOURCE_YAML,
+            dag=self.dag,
+            kubernetes_conn_id="kubernetes_default_kube_config",
+            task_id="test_task_id",
+        )
+
+        op.execute(context)
+        mock_kubernetes_hook.assert_called_once_with()
+
+        mock_create_namespaced_persistent_volume_claim.assert_called_once_with(
+            body=yaml.safe_load(TEST_VALID_RESOURCE_YAML), namespace="default"
+        )
+
+    def test_delete_application_from_yaml(
+        self,
+        context,
+        mock_delete_namespaced_persistent_volume_claim,
+        mock_create_namespaced_persistent_volume_claim,
+        mock_kubernetes_hook,
+    ):
+        op = KubernetesDeleteResourceOperator(
+            yaml_conf=TEST_VALID_RESOURCE_YAML,
+            dag=self.dag,
+            kubernetes_conn_id="kubernetes_default_kube_config",
+            task_id="test_task_id",
+        )
+
+        op.execute(context)
+        mock_kubernetes_hook.assert_called_once_with()
+
+        mock_delete_namespaced_persistent_volume_claim(
+            name="test_pvc",
+            namespace="default",
+            body={
+                "api_version": None,
+                "dry_run": None,
+                "grace_period_seconds": 5,
+                "kind": None,
+                "orphan_dependents": None,
+                "preconditions": None,
+                "propagation_policy": "Background",
+            },
+        )
diff --git a/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py b/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py
new file mode 100644
index 0000000000000..1baaa2b2660a0
--- /dev/null
+++ b/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py
@@ -0,0 +1,97 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+This is an example DAG which uses KubernetesCreateResourceOperator and KubernetesDeleteResourceOperator.
+In this example, we create two tasks which execute sequentially.
+The first task is to create a PVC on Kubernetes cluster.
+and the second task is to delete the PVC.
+"""
+from __future__ import annotations
+
+import os
+from datetime import datetime, timedelta
+
+# [START import_module]
+# The DAG object; we'll need this to instantiate a DAG
+from airflow import DAG
+
+# Operators; we need this to operate!
+from airflow.providers.cncf.kubernetes.operators.resource import (
+    KubernetesCreateResourceOperator,
+    KubernetesDeleteResourceOperator,
+)
+
+# [END import_module]
+
+
+# [START instantiate_dag]
+
+pvc_name = "toto"
+
+pvc_conf = f"""
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: {pvc_name}
+spec:
+  accessModes:
+    - ReadWriteOnce
+  storageClassName: standard-rwo
+  resources:
+    requests:
+      storage: 500Gi
+"""
+
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "example_kubernetes_resource_operator"
+
+
+with DAG(
+    DAG_ID,
+    default_args={"max_active_runs": 1},
+    description="create and delete a PVC in a kubernetes",
+    schedule=timedelta(days=1),
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
+) as dag:
+    # [START KubernetesCreateResourceOperator_DAG]
+    # [START KubernetesDeleteResourceOperator_DAG]
+    t1 = KubernetesCreateResourceOperator(
+        task_id="create_pvc",
+        yaml_conf=pvc_conf,
+    )
+
+    t2 = KubernetesDeleteResourceOperator(
+        task_id="delete_pvc",
+        yaml_conf=pvc_conf,
+    )
+
+    t1 >> t2
+    # [END KubernetesDeleteResourceOperator_DAG]
+    # [END KubernetesCreateResourceOperator_DAG]
+
+    from tests.system.utils.watcher import watcher
+
+    # This test needs watcher in order to properly mark success/failure
+    # when "tearDown" task with trigger rule is part of the DAG
+    list(dag.tasks) >> watcher()
+
+from tests.system.utils import get_test_run  # noqa: E402
+
+# Needed to run the example DAG with pytest (see: tests/system/README.md#run_via_pytest)
+test_run = get_test_run(dag)

From 0105dc25baa865aec67d9adf92f8bb89d8ad02d2 Mon Sep 17 00:00:00 2001
From: raphaelauv <raphaelauv@gmail.com>
Date: Tue, 11 Apr 2023 16:42:47 +0200
Subject: [PATCH 03/10] clean

---
 .../cncf/kubernetes/operators/test_resource.py    | 15 +--------------
 1 file changed, 1 insertion(+), 14 deletions(-)

diff --git a/tests/providers/cncf/kubernetes/operators/test_resource.py b/tests/providers/cncf/kubernetes/operators/test_resource.py
index 5d3b7423d6fcd..b68b9b9327556 100644
--- a/tests/providers/cncf/kubernetes/operators/test_resource.py
+++ b/tests/providers/cncf/kubernetes/operators/test_resource.py
@@ -106,17 +106,4 @@ def test_delete_application_from_yaml(
 
         op.execute(context)
         mock_kubernetes_hook.assert_called_once_with()
-
-        mock_delete_namespaced_persistent_volume_claim(
-            name="test_pvc",
-            namespace="default",
-            body={
-                "api_version": None,
-                "dry_run": None,
-                "grace_period_seconds": 5,
-                "kind": None,
-                "orphan_dependents": None,
-                "preconditions": None,
-                "propagation_policy": "Background",
-            },
-        )
+        mock_delete_namespaced_persistent_volume_claim.assert_called_once()

From a0b301046469ba0ba441fad88a5483cb547ef30e Mon Sep 17 00:00:00 2001
From: raphaelauv <raphaelauv@gmail.com>
Date: Tue, 11 Apr 2023 23:13:55 +0200
Subject: [PATCH 04/10] review

---
 .../cncf/kubernetes/operators/resource.py     |  8 +---
 .../cncf/kubernetes/utils/delete_from.py      | 44 ++++++++++---------
 .../kubernetes/operators/test_resource.py     |  4 +-
 .../kubernetes/example_kubernetes_resource.py |  4 +-
 4 files changed, 29 insertions(+), 31 deletions(-)

diff --git a/airflow/providers/cncf/kubernetes/operators/resource.py b/airflow/providers/cncf/kubernetes/operators/resource.py
index 654d1860d827d..869326ab16db8 100644
--- a/airflow/providers/cncf/kubernetes/operators/resource.py
+++ b/airflow/providers/cncf/kubernetes/operators/resource.py
@@ -71,18 +71,13 @@ def get_namespace(self) -> str:
         if self._namespace:
             return self._namespace
         else:
-            tmp = self.hook.get_namespace()
-            if tmp:
-                return tmp
-            else:
-                return "default"
+            return self.hook.get_namespace() or "default"
 
 
 class KubernetesCreateResourceOperator(KubernetesResourceBaseOperator):
     """Create a resource in a kubernetes."""
 
     def execute(self, context) -> None:
-        print(self.yaml_conf)
         create_from_yaml(
             k8s_client=self.client,
             yaml_objects=[yaml.safe_load(self.yaml_conf)],
@@ -94,7 +89,6 @@ class KubernetesDeleteResourceOperator(KubernetesResourceBaseOperator):
     """Delete a resource in a kubernetes."""
 
     def execute(self, context) -> None:
-        print(self.yaml_conf)
         delete_from_dict(
             k8s_client=self.client,
             yml_document=yaml.safe_load(self.yaml_conf),
diff --git a/airflow/providers/cncf/kubernetes/utils/delete_from.py b/airflow/providers/cncf/kubernetes/utils/delete_from.py
index eb2d235a6f956..543e07a0527d7 100644
--- a/airflow/providers/cncf/kubernetes/utils/delete_from.py
+++ b/airflow/providers/cncf/kubernetes/utils/delete_from.py
@@ -22,6 +22,7 @@
 import re
 
 from kubernetes import client
+from kubernetes.client import ApiClient
 
 DEFAULT_DELETION_BODY = client.V1DeleteOptions(
     propagation_policy="Background",
@@ -30,11 +31,12 @@
 
 
 def delete_from_dict(
-    k8s_client,
-    yml_document,
-    verbose=False,
-    namespace="default",
-    body=None,
+    *,
+    k8s_client: ApiClient,
+    yml_document: dict,
+    verbose: bool = False,
+    namespace: str = "default",
+    body: dict | None = None,
     **kwargs,
 ):
 
@@ -50,9 +52,9 @@ def delete_from_dict(
                 yml_doc["kind"] = kind
             try:
                 _delete_from_yaml_single_item(
-                    k8s_client,
-                    yml_doc,
-                    verbose,
+                    k8s_client=k8s_client,
+                    yml_document=yml_doc,
+                    verbose=verbose,
                     namespace=namespace,
                     body=body,
                     **kwargs,
@@ -63,9 +65,9 @@ def delete_from_dict(
 
         try:
             _delete_from_yaml_single_item(
-                k8s_client,
-                yml_document,
-                verbose,
+                k8s_client=k8s_client,
+                yml_document=yml_document,
+                verbose=verbose,
                 namespace=namespace,
                 body=body,
                 **kwargs,
@@ -78,11 +80,12 @@ def delete_from_dict(
 
 
 def _delete_from_yaml_single_item(
-    k8s_client,
-    yml_document,
-    verbose=False,
-    namespace="default",
-    body=None,
+    *,
+    k8s_client: ApiClient,
+    yml_document: dict,
+    verbose: bool = False,
+    namespace: str = "default",
+    body: dict | None = None,
     **kwargs,
 ):
     if body is None:
@@ -107,19 +110,20 @@ def _delete_from_yaml_single_item(
     kind = re.sub("([a-z0-9])([A-Z])", r"\1_\2", kind).lower()
 
     # Decide which namespace we are going to use for deleting the object
-    # IMPORTANT: Its ignore namespace in args:
+    # IMPORTANT: the docs namespace takes precedence over the namespace in args
     #    create_from_yaml_single_item have same behaviour
     if "namespace" in yml_document["metadata"]:
         namespace = yml_document["metadata"]["namespace"]
     name = yml_document["metadata"]["name"]
 
     # Expect the user to delete namespaced objects more often
+    resp: client.V1Status
     if hasattr(k8s_api, f"delete_namespaced_{kind}"):
-        resp: client.V1Status = getattr(k8s_api, f"delete_namespaced_{kind}")(
+        resp = getattr(k8s_api, f"delete_namespaced_{kind}")(
             name=name, namespace=namespace, body=body, **kwargs
         )
     else:
-        resp: client.V1Status = getattr(k8s_api, f"delete_{kind}")(name=name, body=body, **kwargs)
+        resp = getattr(k8s_api, f"delete_{kind}")(name=name, body=body, **kwargs)
     if verbose:
         print(f"{kind} deleted. status='{str(resp.status)}'")
     return resp
@@ -131,7 +135,7 @@ class FailToDeleteError(Exception):
     handling a yaml file during deletion of the resource.
     """
 
-    def __init__(self, api_exceptions):
+    def __init__(self, api_exceptions: list):
         self.api_exceptions = api_exceptions
 
     def __str__(self):
diff --git a/tests/providers/cncf/kubernetes/operators/test_resource.py b/tests/providers/cncf/kubernetes/operators/test_resource.py
index b68b9b9327556..8b6fad74f3286 100644
--- a/tests/providers/cncf/kubernetes/operators/test_resource.py
+++ b/tests/providers/cncf/kubernetes/operators/test_resource.py
@@ -37,10 +37,10 @@
 spec:
   accessModes:
     - ReadWriteOnce
-  storageClassName: standard-rwo
+  storageClassName: standard
   resources:
     requests:
-      storage: 500Gi
+      storage: 5Gi
 """
 
 
diff --git a/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py b/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py
index 1baaa2b2660a0..a49e99537d9b5 100644
--- a/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py
+++ b/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py
@@ -51,10 +51,10 @@
 spec:
   accessModes:
     - ReadWriteOnce
-  storageClassName: standard-rwo
+  storageClassName: standard
   resources:
     requests:
-      storage: 500Gi
+      storage: 5Gi
 """
 
 ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")

From 5ce94e7a01b90bb2315ebb42c3ffe7ec760a3fd9 Mon Sep 17 00:00:00 2001
From: raphaelauv <raphaelauv@gmail.com>
Date: Sat, 29 Apr 2023 22:06:28 +0200
Subject: [PATCH 05/10] review 2

---
 .../cncf/kubernetes/operators/resource.py     | 12 +--
 .../cncf/kubernetes/utils/delete_from.py      | 50 ++++++----
 .../kubernetes/operators/test_resource.py     | 91 +++++++++++--------
 3 files changed, 88 insertions(+), 65 deletions(-)

diff --git a/airflow/providers/cncf/kubernetes/operators/resource.py b/airflow/providers/cncf/kubernetes/operators/resource.py
index 869326ab16db8..269a2143c819b 100644
--- a/airflow/providers/cncf/kubernetes/operators/resource.py
+++ b/airflow/providers/cncf/kubernetes/operators/resource.py
@@ -18,14 +18,14 @@
 
 from __future__ import annotations
 
+import yaml
 from kubernetes.client import ApiClient
 from kubernetes.utils import create_from_yaml
 
 from airflow.compat.functools import cached_property
 from airflow.models import BaseOperator
 from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
-from airflow.providers.cncf.kubernetes.utils.delete_from import delete_from_dict
-from airflow.utils import yaml
+from airflow.providers.cncf.kubernetes.utils.delete_from import delete_from_yaml
 
 
 class KubernetesResourceBaseOperator(BaseOperator):
@@ -39,7 +39,7 @@ def __init__(
         *,
         yaml_conf: str,
         namespace: str | None = None,
-        kubernetes_conn_id: str | None = "kubernetes_default",
+        kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
         in_cluster: bool | None = None,
         cluster_context: str | None = None,
         config_file: str | None = None,
@@ -80,7 +80,7 @@ class KubernetesCreateResourceOperator(KubernetesResourceBaseOperator):
     def execute(self, context) -> None:
         create_from_yaml(
             k8s_client=self.client,
-            yaml_objects=[yaml.safe_load(self.yaml_conf)],
+            yaml_objects=yaml.safe_load_all(self.yaml_conf),
             namespace=self.get_namespace(),
         )
 
@@ -89,8 +89,8 @@ class KubernetesDeleteResourceOperator(KubernetesResourceBaseOperator):
     """Delete a resource in a kubernetes."""
 
     def execute(self, context) -> None:
-        delete_from_dict(
+        delete_from_yaml(
             k8s_client=self.client,
-            yml_document=yaml.safe_load(self.yaml_conf),
+            yaml_objects=yaml.safe_load_all(self.yaml_conf),
             namespace=self.get_namespace(),
         )
diff --git a/airflow/providers/cncf/kubernetes/utils/delete_from.py b/airflow/providers/cncf/kubernetes/utils/delete_from.py
index 543e07a0527d7..59d1896b78f76 100644
--- a/airflow/providers/cncf/kubernetes/utils/delete_from.py
+++ b/airflow/providers/cncf/kubernetes/utils/delete_from.py
@@ -30,25 +30,14 @@
 )
 
 
-def delete_from_dict(
-    *,
-    k8s_client: ApiClient,
-    yml_document: dict,
-    verbose: bool = False,
-    namespace: str = "default",
-    body: dict | None = None,
-    **kwargs,
-):
-
-    if body is None:
-        body = DEFAULT_DELETION_BODY
-
+def delete_from_dict(k8s_client, data, body, namespace, verbose=False, **kwargs):
     api_exceptions = []
-    if "List" in yml_document["kind"]:
-        kind = yml_document["kind"].replace("List", "")
-        for yml_doc in yml_document["items"]:
+
+    if "List" in data["kind"]:
+        kind = data["kind"].replace("List", "")
+        for yml_doc in data["items"]:
             if kind != "":
-                yml_doc["apiVersion"] = yml_document["apiVersion"]
+                yml_doc["apiVersion"] = data["apiVersion"]
                 yml_doc["kind"] = kind
             try:
                 _delete_from_yaml_single_item(
@@ -66,7 +55,7 @@ def delete_from_dict(
         try:
             _delete_from_yaml_single_item(
                 k8s_client=k8s_client,
-                yml_document=yml_document,
+                yml_document=data,
                 verbose=verbose,
                 namespace=namespace,
                 body=body,
@@ -79,6 +68,29 @@ def delete_from_dict(
         raise FailToDeleteError(api_exceptions)
 
 
+def delete_from_yaml(
+    *,
+    k8s_client: ApiClient,
+    yaml_objects=None,
+    verbose: bool = False,
+    namespace: str = "default",
+    body: dict | None = None,
+    **kwargs,
+):
+    for yml_document in yaml_objects:
+        if yml_document is None:
+            continue
+        else:
+            delete_from_dict(
+                k8s_client=k8s_client,
+                data=yml_document,
+                body=body,
+                namespace=namespace,
+                verbose=verbose,
+                **kwargs,
+            )
+
+
 def _delete_from_yaml_single_item(
     *,
     k8s_client: ApiClient,
@@ -141,5 +153,5 @@ def __init__(self, api_exceptions: list):
     def __str__(self):
         msg = ""
         for api_exception in self.api_exceptions:
-            msg += f"Error from server ({api_exception.reason}):{api_exception.body}"
+            msg += f"Error from server ({api_exception.reason}):{api_exception.body}\n"
         return msg
diff --git a/tests/providers/cncf/kubernetes/operators/test_resource.py b/tests/providers/cncf/kubernetes/operators/test_resource.py
index 8b6fad74f3286..a565e84fc1aa9 100644
--- a/tests/providers/cncf/kubernetes/operators/test_resource.py
+++ b/tests/providers/cncf/kubernetes/operators/test_resource.py
@@ -16,18 +16,17 @@
 # under the License.
 from __future__ import annotations
 
-import json
 from unittest.mock import patch
 
+import pytest
 import yaml
 
 from airflow import DAG
-from airflow.models import Connection
 from airflow.providers.cncf.kubernetes.operators.resource import (
     KubernetesCreateResourceOperator,
     KubernetesDeleteResourceOperator,
 )
-from airflow.utils import db, timezone
+from airflow.utils import timezone
 
 TEST_VALID_RESOURCE_YAML = """
 apiVersion: v1
@@ -43,67 +42,79 @@
       storage: 5Gi
 """
 
+TEST_VALID_LIST_RESOURCE_YAML = """
+apiVersion: v1
+kind: List
+items:
+- apiVersion: v1
+  kind: PersistentVolumeClaim
+  metadata:
+    name: test_pvc_1
+- apiVersion: v1
+  kind: PersistentVolumeClaim
+  metadata:
+    name: test_pvc_2
+"""
+
+HOOK_CLASS = "airflow.providers.cncf.kubernetes.hooks.kubernetes.KubernetesHook"
+
 
-@patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.KubernetesHook.get_conn")
-@patch("kubernetes.client.api.CoreV1Api.create_namespaced_persistent_volume_claim")
-@patch("kubernetes.client.api.CoreV1Api.delete_namespaced_persistent_volume_claim")
 @patch("airflow.utils.context.Context")
-class TestKubernetesCreateOperator:
-    def setup_method(self):
-        db.merge_conn(
-            Connection(
-                conn_id="kubernetes_default_kube_config",
-                conn_type="kubernetes",
-                extra=json.dumps({}),
-            )
-        )
+class TestKubernetesXResourceOperator:
+    @pytest.fixture(autouse=True)
+    def setup_tests(self, dag_maker):
+        self._default_client_patch = patch(f"{HOOK_CLASS}._get_default_client")
+        self._default_client_mock = self._default_client_patch.start()
 
-        db.merge_conn(
-            Connection(
-                conn_id="kubernetes_with_namespace",
-                conn_type="kubernetes",
-                extra=json.dumps({"namespace": "mock_namespace"}),
-            )
-        )
+        yield
+
+        patch.stopall()
 
+    def setup_method(self):
         args = {"owner": "airflow", "start_date": timezone.datetime(2020, 2, 1)}
         self.dag = DAG("test_dag_id", default_args=args)
 
-    def test_create_application_from_yaml(
-        self,
-        context,
-        mock_delete_namespaced_persistent_volume_claim,
-        mock_create_namespaced_persistent_volume_claim,
-        mock_kubernetes_hook,
-    ):
+    @patch("kubernetes.client.api.CoreV1Api.create_namespaced_persistent_volume_claim")
+    def test_create_application_from_yaml(self, mock_create_namespaced_persistent_volume_claim, context):
         op = KubernetesCreateResourceOperator(
             yaml_conf=TEST_VALID_RESOURCE_YAML,
             dag=self.dag,
-            kubernetes_conn_id="kubernetes_default_kube_config",
+            kubernetes_conn_id="kubernetes_default",
             task_id="test_task_id",
         )
 
         op.execute(context)
-        mock_kubernetes_hook.assert_called_once_with()
 
         mock_create_namespaced_persistent_volume_claim.assert_called_once_with(
             body=yaml.safe_load(TEST_VALID_RESOURCE_YAML), namespace="default"
         )
 
-    def test_delete_application_from_yaml(
-        self,
-        context,
-        mock_delete_namespaced_persistent_volume_claim,
-        mock_create_namespaced_persistent_volume_claim,
-        mock_kubernetes_hook,
+    @patch("kubernetes.client.api.CoreV1Api.delete_namespaced_persistent_volume_claim")
+    def test_single_delete_application_from_yaml(
+        self, mock_delete_namespaced_persistent_volume_claim, context
     ):
         op = KubernetesDeleteResourceOperator(
             yaml_conf=TEST_VALID_RESOURCE_YAML,
             dag=self.dag,
-            kubernetes_conn_id="kubernetes_default_kube_config",
+            kubernetes_conn_id="kubernetes_default",
             task_id="test_task_id",
         )
 
         op.execute(context)
-        mock_kubernetes_hook.assert_called_once_with()
-        mock_delete_namespaced_persistent_volume_claim.assert_called_once()
+
+        mock_delete_namespaced_persistent_volume_claim.assert_called()
+
+    @patch("kubernetes.client.api.CoreV1Api.delete_namespaced_persistent_volume_claim")
+    def test_multi_delete_application_from_yaml(
+        self, mock_delete_namespaced_persistent_volume_claim, context
+    ):
+        op = KubernetesDeleteResourceOperator(
+            yaml_conf=TEST_VALID_LIST_RESOURCE_YAML,
+            dag=self.dag,
+            kubernetes_conn_id="kubernetes_default",
+            task_id="test_task_id",
+        )
+
+        op.execute(context)
+
+        mock_delete_namespaced_persistent_volume_claim.assert_called()

From c4827b61ba98cac0d45e9d599dd00299764e66eb Mon Sep 17 00:00:00 2001
From: raphaelauv <raphaelauv@gmail.com>
Date: Fri, 5 May 2023 04:18:31 +0200
Subject: [PATCH 06/10] remove base_class from missing examples

---
 tests/always/test_project_structure.py | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/tests/always/test_project_structure.py b/tests/always/test_project_structure.py
index 75444994876a7..b17e285bdb33a 100644
--- a/tests/always/test_project_structure.py
+++ b/tests/always/test_project_structure.py
@@ -430,10 +430,6 @@ class TestCncfProviderProjectStructure(ExampleCoverageTest):
     }
     BASE_CLASSES = {"airflow.providers.cncf.kubernetes.operators.resource.KubernetesResourceBaseOperator"}
 
-    MISSING_EXAMPLES_FOR_CLASSES = {
-        "airflow.providers.cncf.kubernetes.operators.resource.KubernetesResourceBaseOperator"
-    }
-
 
 class TestSlackProviderProjectStructure(ExampleCoverageTest):
     PROVIDER = "slack"

From ace319f675b747c94dab696bf8659e5a51430ddf Mon Sep 17 00:00:00 2001
From: raphaelauv <raphaelauv@gmail.com>
Date: Fri, 26 May 2023 01:10:50 +0200
Subject: [PATCH 07/10] review 3

---
 .../cncf/kubernetes/operators/resource.py     | 19 ++++++++++++++++++-
 1 file changed, 18 insertions(+), 1 deletion(-)

diff --git a/airflow/providers/cncf/kubernetes/operators/resource.py b/airflow/providers/cncf/kubernetes/operators/resource.py
index 269a2143c819b..0d19f107ab021 100644
--- a/airflow/providers/cncf/kubernetes/operators/resource.py
+++ b/airflow/providers/cncf/kubernetes/operators/resource.py
@@ -27,9 +27,26 @@
 from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
 from airflow.providers.cncf.kubernetes.utils.delete_from import delete_from_yaml
 
+__all__ = ["KubernetesCreateResourceOperator", "KubernetesDeleteResourceOperator"]
+
 
 class KubernetesResourceBaseOperator(BaseOperator):
-    """Abstract base class for all Kubernetes Resource operators."""
+    """
+    Abstract base class for all Kubernetes Resource operators.
+
+    :param yaml_conf: string. Contains the kubernetes resources to Create or Delete
+    :param namespace: string. Contains the namespace to create all resources inside.
+        The namespace must preexist otherwise the resource creation will fail.
+        If the API object in the yaml file already contains a namespace definition then
+        this parameter has no effect.
+    :param kubernetes_conn_id: The :ref:`kubernetes connection id <howto/connection:kubernetes>`
+        for the Kubernetes cluster.
+    :param in_cluster: run kubernetes client with in_cluster configuration.
+    :param cluster_context: context that points to kubernetes cluster.
+        Ignored when in_cluster is True. If None, current-context is used.
+    :param config_file: The path to the Kubernetes config file. (templated)
+        If not specified, default value is ``~/.kube/config``
+    """
 
     template_fields = ("yaml_conf",)
     template_fields_renderers = {"yaml_conf": "yaml"}

From a3f3103020425685a7b98b2c2cc239a2ab0466d0 Mon Sep 17 00:00:00 2001
From: raphaelauv <raphaelauv@gmail.com>
Date: Fri, 26 May 2023 12:08:37 +0200
Subject: [PATCH 08/10] review 4

---
 .../providers/cncf/kubernetes/operators/resource.py | 13 +------------
 1 file changed, 1 insertion(+), 12 deletions(-)

diff --git a/airflow/providers/cncf/kubernetes/operators/resource.py b/airflow/providers/cncf/kubernetes/operators/resource.py
index 0d19f107ab021..7bcc302148ee6 100644
--- a/airflow/providers/cncf/kubernetes/operators/resource.py
+++ b/airflow/providers/cncf/kubernetes/operators/resource.py
@@ -57,17 +57,11 @@ def __init__(
         yaml_conf: str,
         namespace: str | None = None,
         kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
-        in_cluster: bool | None = None,
-        cluster_context: str | None = None,
-        config_file: str | None = None,
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
         self._namespace = namespace
         self.kubernetes_conn_id = kubernetes_conn_id
-        self.in_cluster = in_cluster
-        self.cluster_context = cluster_context
-        self.config_file = config_file
         self.yaml_conf = yaml_conf
 
     @cached_property
@@ -76,12 +70,7 @@ def client(self) -> ApiClient:
 
     @cached_property
     def hook(self) -> KubernetesHook:
-        hook = KubernetesHook(
-            conn_id=self.kubernetes_conn_id,
-            in_cluster=self.in_cluster,
-            config_file=self.config_file,
-            cluster_context=self.cluster_context,
-        )
+        hook = KubernetesHook(conn_id=self.kubernetes_conn_id)
         return hook
 
     def get_namespace(self) -> str:

From 279c11b959cb760edfc50dfc4cf4e572b31fe625 Mon Sep 17 00:00:00 2001
From: raphaelauv <raphaelauv@gmail.com>
Date: Sun, 4 Jun 2023 15:22:06 +0200
Subject: [PATCH 09/10] review 5

---
 .../cncf/kubernetes/example_kubernetes_resource.py | 14 --------------
 1 file changed, 14 deletions(-)

diff --git a/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py b/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py
index a49e99537d9b5..7be0a8099106b 100644
--- a/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py
+++ b/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py
@@ -26,21 +26,12 @@
 import os
 from datetime import datetime, timedelta
 
-# [START import_module]
-# The DAG object; we'll need this to instantiate a DAG
 from airflow import DAG
-
-# Operators; we need this to operate!
 from airflow.providers.cncf.kubernetes.operators.resource import (
     KubernetesCreateResourceOperator,
     KubernetesDeleteResourceOperator,
 )
 
-# [END import_module]
-
-
-# [START instantiate_dag]
-
 pvc_name = "toto"
 
 pvc_conf = f"""
@@ -60,7 +51,6 @@
 ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
 DAG_ID = "example_kubernetes_resource_operator"
 
-
 with DAG(
     DAG_ID,
     default_args={"max_active_runs": 1},
@@ -69,8 +59,6 @@
     start_date=datetime(2021, 1, 1),
     catchup=False,
 ) as dag:
-    # [START KubernetesCreateResourceOperator_DAG]
-    # [START KubernetesDeleteResourceOperator_DAG]
     t1 = KubernetesCreateResourceOperator(
         task_id="create_pvc",
         yaml_conf=pvc_conf,
@@ -82,8 +70,6 @@
     )
 
     t1 >> t2
-    # [END KubernetesDeleteResourceOperator_DAG]
-    # [END KubernetesCreateResourceOperator_DAG]
 
     from tests.system.utils.watcher import watcher
 

From a336d6372a5886049bd56c4efe22da4449f7be0a Mon Sep 17 00:00:00 2001
From: raphaelauv <raphaelauv@gmail.com>
Date: Sun, 4 Jun 2023 16:38:37 +0200
Subject: [PATCH 10/10] fix import

---
 airflow/providers/cncf/kubernetes/operators/resource.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/airflow/providers/cncf/kubernetes/operators/resource.py b/airflow/providers/cncf/kubernetes/operators/resource.py
index 7bcc302148ee6..9515e5b7e5e56 100644
--- a/airflow/providers/cncf/kubernetes/operators/resource.py
+++ b/airflow/providers/cncf/kubernetes/operators/resource.py
@@ -18,11 +18,12 @@
 
 from __future__ import annotations
 
+from functools import cached_property
+
 import yaml
 from kubernetes.client import ApiClient
 from kubernetes.utils import create_from_yaml
 
-from airflow.compat.functools import cached_property
 from airflow.models import BaseOperator
 from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
 from airflow.providers.cncf.kubernetes.utils.delete_from import delete_from_yaml
