Revive PR #1392: solves #940 and update to latest

(original by @DiptoChakrabarty)

- Implement delete_from_yaml functionality
- Combine create_from_yaml and delete_from_yaml into yaml_processor, preserving existing signatures for create

Author: kocarba

kubernetes/e2e_test/test_utils.py

@@ -17,6 +17,7 @@
 from os import path
 
 import yaml
+import time
 
 from kubernetes import client, utils
 from kubernetes.client.rest import ApiException
@@ -36,6 +37,15 @@ def setUpClass(cls):
         body = client.V1Namespace(
             metadata=client.V1ObjectMeta(
                 name=cls.test_namespace))
+
+        # Delete the namespace if it already exists
+        try:
+            core_v1.delete_namespace(name=cls.test_namespace)
+            time.sleep(10)  # Wait for the namespace to be deleted
+        except ApiException as e:
+            if e.status != 404:
+                raise
+
         core_v1.create_namespace(body=body)
 
     @classmethod
@@ -51,15 +61,17 @@ def test_create_apps_deployment_from_yaml(self):
         Should be able to create an apps/v1 deployment.
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "apps-deployment.yaml")
         app_api = client.AppsV1Api(k8s_client)
         dep = app_api.read_namespaced_deployment(name="nginx-app",
                                                  namespace="default")
         self.assertIsNotNone(dep)
         self.assertEqual("nginx-app", dep.metadata.name)
-        self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
-        self.assertEqual(80, dep.spec.template.spec.containers[0].ports[0].container_port)
+        self.assertEqual(
+            "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+        self.assertEqual(
+            80, dep.spec.template.spec.containers[0].ports[0].container_port)
         self.assertEqual("nginx", dep.spec.template.spec.containers[0].name)
         self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
         self.assertEqual(3, dep.spec.replicas)
@@ -79,28 +91,34 @@ def test_create_apps_deployment_from_yaml_with_apply_is_idempotent(self):
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
         try:
-            utils.create_from_yaml(
+            utils.process_from_yaml(
                 k8s_client, self.path_prefix + "apps-deployment.yaml")
             app_api = client.AppsV1Api(k8s_client)
             dep = app_api.read_namespaced_deployment(name="nginx-app",
-                                                    namespace="default")
+                                                     namespace="default")
             self.assertIsNotNone(dep)
             self.assertEqual("nginx-app", dep.metadata.name)
-            self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
-            self.assertEqual(80, dep.spec.template.spec.containers[0].ports[0].container_port)
-            self.assertEqual("nginx", dep.spec.template.spec.containers[0].name)
+            self.assertEqual(
+                "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+            self.assertEqual(
+                80, dep.spec.template.spec.containers[0].ports[0].container_port)
+            self.assertEqual(
+                "nginx", dep.spec.template.spec.containers[0].name)
             self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
             self.assertEqual(3, dep.spec.replicas)
 
-            utils.create_from_yaml(
+            utils.process_from_yaml(
                 k8s_client, self.path_prefix + "apps-deployment.yaml", apply=True)
             dep = app_api.read_namespaced_deployment(name="nginx-app",
-                                                    namespace="default")
+                                                     namespace="default")
             self.assertIsNotNone(dep)
             self.assertEqual("nginx-app", dep.metadata.name)
-            self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
-            self.assertEqual(80, dep.spec.template.spec.containers[0].ports[0].container_port)
-            self.assertEqual("nginx", dep.spec.template.spec.containers[0].name)
+            self.assertEqual(
+                "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+            self.assertEqual(
+                80, dep.spec.template.spec.containers[0].ports[0].container_port)
+            self.assertEqual(
+                "nginx", dep.spec.template.spec.containers[0].name)
             self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
             self.assertEqual(3, dep.spec.replicas)
         except Exception as e:
@@ -123,7 +141,7 @@ def test_create_apps_deployment_from_yaml_object(self):
         _path = self.path_prefix + "apps-deployment.yaml"
         with open(path.abspath(_path)) as f:
             yaml_objects = yaml.safe_load_all(f)
-            utils.create_from_yaml(
+            utils.process_from_yaml(
                 k8s_client,
                 yaml_objects=yaml_objects,
             )
@@ -132,8 +150,10 @@ def test_create_apps_deployment_from_yaml_object(self):
                                                  namespace="default")
         self.assertIsNotNone(dep)
         self.assertEqual("nginx-app", dep.metadata.name)
-        self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
-        self.assertEqual(80, dep.spec.template.spec.containers[0].ports[0].container_port)
+        self.assertEqual(
+            "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+        self.assertEqual(
+            80, dep.spec.template.spec.containers[0].ports[0].container_port)
         self.assertEqual("nginx", dep.spec.template.spec.containers[0].name)
         self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
         self.assertEqual(3, dep.spec.replicas)
@@ -154,15 +174,17 @@ def test_create_apps_deployment_from_yaml_obj(self):
 
         yml_obj["metadata"]["name"] = "nginx-app-3"
 
-        utils.create_from_dict(k8s_client, yml_obj)
+        utils.process_from_dict(k8s_client, yml_obj)
 
         app_api = client.AppsV1Api(k8s_client)
         dep = app_api.read_namespaced_deployment(name="nginx-app-3",
                                                  namespace="default")
         self.assertIsNotNone(dep)
         self.assertEqual("nginx-app-3", dep.metadata.name)
-        self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
-        self.assertEqual(80, dep.spec.template.spec.containers[0].ports[0].container_port)
+        self.assertEqual(
+            "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+        self.assertEqual(
+            80, dep.spec.template.spec.containers[0].ports[0].container_port)
         self.assertEqual("nginx", dep.spec.template.spec.containers[0].name)
         self.assertEqual("nginx", dep.spec.template.metadata.labels["app"])
         self.assertEqual(3, dep.spec.replicas)
@@ -176,7 +198,7 @@ def test_create_pod_from_yaml(self):
         Should be able to create a pod.
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "core-pod.yaml")
         core_api = client.CoreV1Api(k8s_client)
         pod = core_api.read_namespaced_pod(name="myapp-pod",
@@ -195,7 +217,7 @@ def test_create_service_from_yaml(self):
         Should be able to create a service.
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "core-service.yaml")
         core_api = client.CoreV1Api(k8s_client)
         svc = core_api.read_namespaced_service(name="my-service",
@@ -213,7 +235,7 @@ def test_create_namespace_from_yaml(self):
         Should be able to create a namespace.
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "core-namespace.yaml")
         core_api = client.CoreV1Api(k8s_client)
         nmsp = core_api.read_namespace(name="development")
@@ -228,7 +250,7 @@ def test_create_rbac_role_from_yaml(self):
         Should be able to create a rbac role.
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "rbac-role.yaml")
         rbac_api = client.RbacAuthorizationV1Api(k8s_client)
         rbac_role = rbac_api.read_namespaced_role(
@@ -245,7 +267,7 @@ def test_create_rbac_role_from_yaml_with_verbose_enabled(self):
         Should be able to create a rbac role with verbose enabled.
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "rbac-role.yaml", verbose=True)
         rbac_api = client.RbacAuthorizationV1Api(k8s_client)
         rbac_role = rbac_api.read_namespaced_role(
@@ -263,9 +285,9 @@ def test_create_deployment_non_default_namespace_from_yaml(self):
         and then create a deployment in the just-created namespace.
         """
         k8s_client = client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "dep-namespace.yaml")
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "dep-deployment.yaml")
         core_api = client.CoreV1Api(k8s_client)
         ext_api = client.AppsV1Api(k8s_client)
@@ -288,7 +310,7 @@ def test_create_apiservice_from_yaml_with_conflict(self):
         fail due to conflict.
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "api-service.yaml")
         reg_api = client.ApiregistrationV1Api(k8s_client)
         svc = reg_api.read_api_service(
@@ -298,8 +320,8 @@ def test_create_apiservice_from_yaml_with_conflict(self):
         self.assertEqual("wardle.k8s.io", svc.spec.group)
         self.assertEqual("v1alpha1", svc.spec.version)
 
-        with self.assertRaises(utils.FailToCreateError) as cm:
-            utils.create_from_yaml(
+        with self.assertRaises(utils.FailToProcessError) as cm:
+            utils.process_from_yaml(
                 k8s_client, "kubernetes/e2e_test/test_yaml/api-service.yaml")
         exp_error = ('Error from server (Conflict): '
                      '{"kind":"Status","apiVersion":"v1","metadata":{},'
@@ -323,7 +345,7 @@ def test_create_general_list_from_yaml(self):
         from a kind: List yaml file
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "list.yaml")
         core_api = client.CoreV1Api(k8s_client)
         ext_api = client.AppsV1Api(k8s_client)
@@ -337,7 +359,8 @@ def test_create_general_list_from_yaml(self):
                                                  namespace="default")
         self.assertIsNotNone(dep)
         self.assertEqual("list-deployment-test", dep.metadata.name)
-        self.assertEqual("nginx:1.15.4", dep.spec.template.spec.containers[0].image)
+        self.assertEqual(
+            "nginx:1.15.4", dep.spec.template.spec.containers[0].image)
         self.assertEqual(1, dep.spec.replicas)
 
         core_api.delete_namespaced_service(name="list-service-test",
@@ -351,7 +374,7 @@ def test_create_namespace_list_from_yaml(self):
         from a kind: NamespaceList yaml file
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "namespace-list.yaml")
         core_api = client.CoreV1Api(k8s_client)
         nmsp_1 = core_api.read_namespace(name="mock-1")
@@ -373,8 +396,8 @@ def test_create_implicit_service_list_from_yaml_with_conflict(self):
         json file that implicitly indicates the kind of individual objects
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        with self.assertRaises(utils.FailToCreateError):
-            utils.create_from_yaml(
+        with self.assertRaises(utils.FailToProcessError):
+            utils.process_from_yaml(
                 k8s_client, self.path_prefix + "implicit-svclist.json")
         core_api = client.CoreV1Api(k8s_client)
         svc_3 = core_api.read_namespaced_service(name="mock-3",
@@ -402,7 +425,7 @@ def test_create_multi_resource_from_directory(self):
         from a directory
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_directory(
+        utils.process_from_directory(
             k8s_client, self.path_prefix + "multi-resource/")
         core_api = client.CoreV1Api(k8s_client)
         svc = core_api.read_namespaced_service(name="mock",
@@ -419,8 +442,10 @@ def test_create_multi_resource_from_directory(self):
         self.assertEqual("mock", ctr.spec.template.metadata.labels["app"])
         self.assertEqual("mock", ctr.spec.selector["app"])
         self.assertEqual(1, ctr.spec.replicas)
-        self.assertEqual("k8s.gcr.io/pause:2.0", ctr.spec.template.spec.containers[0].image)
-        self.assertEqual("mock-container", ctr.spec.template.spec.containers[0].name)
+        self.assertEqual("k8s.gcr.io/pause:2.0",
+                         ctr.spec.template.spec.containers[0].image)
+        self.assertEqual("mock-container",
+                         ctr.spec.template.spec.containers[0].name)
 
         core_api.delete_namespaced_replication_controller(
             name="mock", namespace="default", propagation_policy="Background")
@@ -435,7 +460,7 @@ def test_create_from_multi_resource_yaml(self):
         from a multi-resource yaml file
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "multi-resource.yaml")
         core_api = client.CoreV1Api(k8s_client)
         svc = core_api.read_namespaced_service(name="mock",
@@ -452,8 +477,10 @@ def test_create_from_multi_resource_yaml(self):
         self.assertEqual("mock", ctr.spec.template.metadata.labels["app"])
         self.assertEqual("mock", ctr.spec.selector["app"])
         self.assertEqual(1, ctr.spec.replicas)
-        self.assertEqual("k8s.gcr.io/pause:2.0", ctr.spec.template.spec.containers[0].image)
-        self.assertEqual("mock-container", ctr.spec.template.spec.containers[0].name)
+        self.assertEqual("k8s.gcr.io/pause:2.0",
+                         ctr.spec.template.spec.containers[0].image)
+        self.assertEqual("mock-container",
+                         ctr.spec.template.spec.containers[0].name)
 
         core_api.delete_namespaced_replication_controller(
             name="mock", namespace="default", propagation_policy="Background")
@@ -466,7 +493,7 @@ def test_create_from_list_in_multi_resource_yaml(self):
         specified in the multi-resource file
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "multi-resource-with-list.yaml")
         core_api = client.CoreV1Api(k8s_client)
         app_api = client.AppsV1Api(k8s_client)
@@ -508,7 +535,7 @@ def test_create_from_multi_resource_yaml_with_conflict(self):
         Should raise an exception for failure to create the same service twice.
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "yaml-conflict-first.yaml")
         core_api = client.CoreV1Api(k8s_client)
         svc = core_api.read_namespaced_service(name="mock-2",
@@ -519,8 +546,8 @@ def test_create_from_multi_resource_yaml_with_conflict(self):
         self.assertEqual("mock-2", svc.spec.selector["app"])
         self.assertEqual(99, svc.spec.ports[0].port)
 
-        with self.assertRaises(utils.FailToCreateError) as cm:
-            utils.create_from_yaml(
+        with self.assertRaises(utils.FailToProcessError) as cm:
+            utils.process_from_yaml(
                 k8s_client, self.path_prefix + "yaml-conflict-multi.yaml")
         exp_error = ('Error from server (Conflict): {"kind":"Status",'
                      '"apiVersion":"v1","metadata":{},"status":"Failure",'
@@ -544,8 +571,8 @@ def test_create_from_multi_resource_yaml_with_multi_conflicts(self):
         Should raise an exception that contains two error messages.
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        with self.assertRaises(utils.FailToCreateError) as cm:
-            utils.create_from_yaml(
+        with self.assertRaises(utils.FailToProcessError) as cm:
+            utils.process_from_yaml(
                 k8s_client, self.path_prefix + "triple-nginx.yaml")
         exp_error = ('Error from server (Conflict): {"kind":"Status",'
                      '"apiVersion":"v1","metadata":{},"status":"Failure",'
@@ -570,7 +597,7 @@ def test_create_namespaced_apps_deployment_from_yaml(self):
                 in a test namespace.
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "apps-deployment.yaml",
             namespace=self.test_namespace)
         app_api = client.AppsV1Api(k8s_client)
@@ -587,7 +614,7 @@ def test_create_from_list_in_multi_resource_yaml_namespaced(self):
         specified in the multi-resource file in a test namespace
         """
         k8s_client = client.api_client.ApiClient(configuration=self.config)
-        utils.create_from_yaml(
+        utils.process_from_yaml(
             k8s_client, self.path_prefix + "multi-resource-with-list.yaml",
             namespace=self.test_namespace)
         core_api = client.CoreV1Api(k8s_client)
@@ -608,6 +635,254 @@ def test_create_from_list_in_multi_resource_yaml_namespaced(self):
         app_api.delete_namespaced_deployment(
             name="mock", namespace=self.test_namespace, body={})
 
+    def test_delete_namespace_from_yaml(self):
+        """
+        Should be able to delete a namespace
+        Create namespace from file first and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "core-namespace.yaml")
+        core_api = client.CoreV1Api(k8s_client)
+        nmsp = core_api.read_namespace(name="development")
+        self.assertIsNotNone(nmsp)
+        """
+        Delete namespace from yaml
+        """
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "core-namespace.yaml", action="delete")
+        time.sleep(10)
+        namespace_status = False
+        try:
+            nmsp = core_api.read_namespace(name="development")
+            namespace_status = True
+        except Exception as e:
+            self.assertFalse(namespace_status)
+        self.assertFalse(namespace_status)
+
+    def test_delete_apps_deployment_from_yaml(self):
+        """
+        Should delete a deployment
+        First create deployment from file and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "apps-deployment.yaml")
+        app_api = client.AppsV1Api(k8s_client)
+        dep = app_api.read_namespaced_deployment(name="nginx-app",
+                                                 namespace="default")
+        self.assertIsNotNone(dep)
+        """
+        Deployment should be created
+        Now delete deployment using delete_from_yaml method
+        """
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "apps-deployment.yaml", action="delete")
+        deployment_status = False
+        time.sleep(10)
+        try:
+            response = app_api.read_namespaced_deployment(
+                name="nginx-app", namespace="default")
+            deployment_status = True
+        except Exception as e:
+            self.assertFalse(deployment_status)
+
+        self.assertFalse(deployment_status)
+
+    def test_delete_service_from_yaml(self):
+        """
+        Should be able to delete a service
+        Create service from yaml first and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "core-service.yaml")
+        core_api = client.CoreV1Api(k8s_client)
+        svc = core_api.read_namespaced_service(name="my-service",
+                                               namespace="default")
+        self.assertIsNotNone(svc)
+        """
+        Delete service from yaml
+        """
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "core-service.yaml", action="delete")
+        service_status = False
+        time.sleep(10)
+        try:
+            response = core_api.read_namespaced_service(
+                name="my-service", namespace="default")
+            service_status = True
+        except Exception as e:
+            self.assertFalse(service_status)
+        self.assertFalse(service_status)
+
+    def test_delete_pod_from_yaml(self):
+        """
+        Should be able to delete pod
+        Create pod from file first and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "core-pod.yaml")
+        core_api = client.CoreV1Api(k8s_client)
+        pod = core_api.read_namespaced_pod(name="myapp-pod",
+                                           namespace="default")
+        self.assertIsNotNone(pod)
+        """
+        Delete pod using delete_from_yaml
+        """
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "core-pod.yaml", action="delete")
+        time.sleep(10)
+        pod_status = False
+        try:
+            response = core_api.read_namespaced_pod(name="myapp-pod",
+                                                    namespace="default")
+            pod_status = True
+        except Exception as e:
+            self.assertFalse(pod_status)
+        self.assertFalse(pod_status)
+
+    def test_delete_rbac_role_from_yaml(self):
+        """
+        Should be able to delete rbac role
+        Create rbac role from file first and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "rbac-role.yaml")
+        rbac_api = client.RbacAuthorizationV1Api(k8s_client)
+        rbac_role = rbac_api.read_namespaced_role(
+            name="pod-reader", namespace="default")
+        self.assertIsNotNone(rbac_role)
+        """
+        Delete rbac role from yaml
+        """
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "rbac-role.yaml", action="delete")
+        rbac_role_status = False
+        time.sleep(10)
+        try:
+            response = rbac_api.read_namespaced_role(
+                name="pod-reader", namespace="default")
+            rbac_role_status = True
+        except Exception as e:
+            self.assertFalse(rbac_role_status)
+        self.assertFalse(rbac_role_status)
+
+    def test_delete_rbac_role_from_yaml_with_verbose_enabled(self):
+        """
+        Should delete a rbac role with verbose enabled
+        Create rbac role with verbose enabled and ensure it is created
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "rbac-role.yaml", verbose=True)
+        rbac_api = client.RbacAuthorizationV1Api(k8s_client)
+        rbac_role = rbac_api.read_namespaced_role(
+            name="pod-reader", namespace="default")
+        self.assertIsNotNone(rbac_role)
+        """
+        Delete the rbac role from yaml
+        """
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "rbac-role.yaml", verbose=True, action="delete")
+
+        rbac_role_status = False
+        time.sleep(10)
+        try:
+            response = rbac_api.read_namespaced_role(
+                name="pod-reader", namespace="default")
+            rbac_role_status = True
+        except Exception as e:
+            self.assertFalse(rbac_role_status)
+        self.assertFalse(rbac_role_status)
+
+    # Deletion Tests for multi resource objects in yaml files
+
+    def test_delete_from_multi_resource_yaml(self):
+        """
+        Should be able to delete service and replication controller
+        from the multi resource yaml files
+        Create the resources first and ensure they exist
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "multi-resource.yaml")
+        core_api = client.CoreV1Api(k8s_client)
+        svc = core_api.read_namespaced_service(name="mock",
+                                               namespace="default")
+        self.assertIsNotNone(svc)
+        ctr = core_api.read_namespaced_replication_controller(
+            name="mock", namespace="default")
+        self.assertIsNotNone(ctr)
+        """
+        Delete service and replication controller using yaml file
+        """
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "multi-resource.yaml", action="delete")
+        svc_status = False
+        replication_status = False
+        time.sleep(10)
+        try:
+            resp_svc = core_api.read_namespaced_service(name="mock",
+                                                        namespace="default")
+            svc_status = True
+            resp_repl = core_api.read_namespaced_replication_controller(
+                name="mock", namespace="default")
+            repl_status = True
+        except Exception as e:
+            self.assertFalse(svc_status)
+            self.assertFalse(replication_status)
+        self.assertFalse(svc_status)
+        self.assertFalse(replication_status)
+
+    def test_delete_from_list_in_multi_resource_yaml(self):
+        """
+        Should delete the items in PodList and the deployment in the yaml file
+        Create the items first and ensure they exist
+        """
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "multi-resource-with-list.yaml")
+        core_api = client.CoreV1Api(k8s_client)
+        app_api = client.AppsV1Api(k8s_client)
+        pod_0 = core_api.read_namespaced_pod(
+            name="mock-pod-0", namespace="default")
+        self.assertIsNotNone(pod_0)
+        pod_1 = core_api.read_namespaced_pod(
+            name="mock-pod-1", namespace="default")
+        self.assertIsNotNone(pod_1)
+        dep = app_api.read_namespaced_deployment(
+            name="mock", namespace="default")
+        self.assertIsNotNone(dep)
+        """
+        Delete the PodList and Deployment using the yaml file
+        """
+        utils.process_from_yaml(
+            k8s_client, self.path_prefix + "multi-resource-with-list.yaml", action="delete")
+        time.sleep(10)
+        pod0_status = False
+        pod1_status = False
+        deploy_status = False
+        try:
+            core_api.read_namespaced_pod(
+                name="mock-pod-0", namespace="default")
+            core_api.read_namespaced_pod(
+                name="mock-pod-1", namespace="default")
+            app_api.read_namespaced_deployment(
+                name="mock", namespace="default")
+            pod0_status = True
+            pod1_status = True
+            deploy_status = True
+        except Exception as e:
+            self.assertFalse(pod0_status)
+            self.assertFalse(pod1_status)
+            self.assertFalse(deploy_status)
+        self.assertFalse(pod0_status)
+        self.assertFalse(pod1_status)
+        self.assertFalse(deploy_status)
+
 
 class TestUtilsUnitTests(unittest.TestCase):
 
@@ -624,7 +899,8 @@ def test_parse_quantity(self):
         self.assertRaises(
             ValueError, lambda: quantity.parse_quantity("1000ki")
         )
-        self.assertRaises(ValueError, lambda: quantity.parse_quantity("1000foo"))
+        self.assertRaises(
+            ValueError, lambda: quantity.parse_quantity("1000foo"))
         self.assertRaises(ValueError, lambda: quantity.parse_quantity("foo"))
 
         # == no suffix ==
@@ -641,14 +917,18 @@ def test_parse_quantity(self):
         self.assertEqual(quantity.parse_quantity("0.5Ki"), Decimal(512))
 
         # == base 1000 ==
-        self.assertAlmostEqual(quantity.parse_quantity("1n"), Decimal(0.000_000_001))
-        self.assertAlmostEqual(quantity.parse_quantity("1u"), Decimal(0.000_001))
+        self.assertAlmostEqual(quantity.parse_quantity(
+            "1n"), Decimal(0.000_000_001))
+        self.assertAlmostEqual(
+            quantity.parse_quantity("1u"), Decimal(0.000_001))
         self.assertAlmostEqual(quantity.parse_quantity("1m"), Decimal(0.001))
         self.assertEqual(quantity.parse_quantity("1k"), Decimal(1_000))
         self.assertEqual(quantity.parse_quantity("1M"), Decimal(1_000_000))
         self.assertEqual(quantity.parse_quantity("1G"), Decimal(1_000_000_000))
-        self.assertEqual(quantity.parse_quantity("1T"), Decimal(1_000_000_000_000))
-        self.assertEqual(quantity.parse_quantity("1P"), Decimal(1_000_000_000_000_000))
+        self.assertEqual(quantity.parse_quantity(
+            "1T"), Decimal(1_000_000_000_000))
+        self.assertEqual(quantity.parse_quantity(
+            "1P"), Decimal(1_000_000_000_000_000))
         self.assertEqual(
             quantity.parse_quantity("1E"), Decimal(1_000_000_000_000_000_000))
         self.assertEqual(quantity.parse_quantity("1000k"), Decimal(1_000_000))
@@ -671,17 +951,25 @@ def test_format_quantity(self):
 
         # == no suffix ==
         self.assertEqual(quantity.format_quantity(Decimal(1_000), ""), "1000")
-        self.assertEqual(quantity.format_quantity(Decimal(1_000), None), "1000")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(1_000), None), "1000")
 
         # == base 1024 ==
         self.assertEqual(quantity.format_quantity(Decimal(1024), "Ki"), "1Ki")
-        self.assertEqual(quantity.format_quantity(Decimal(1024**2), "Mi"), "1Mi")
-        self.assertEqual(quantity.format_quantity(Decimal(1024**3), "Gi"), "1Gi")
-        self.assertEqual(quantity.format_quantity(Decimal(1024**4), "Ti"), "1Ti")
-        self.assertEqual(quantity.format_quantity(Decimal(1024**5), "Pi"), "1Pi")
-        self.assertEqual(quantity.format_quantity(Decimal(1024**6), "Ei"), "1Ei")
-        self.assertEqual(quantity.format_quantity(Decimal(1024**2), "Ki"), "1024Ki")
-        self.assertEqual(quantity.format_quantity(Decimal((1024**3) / 2), "Gi"), "0.5Gi")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(1024**2), "Mi"), "1Mi")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(1024**3), "Gi"), "1Gi")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(1024**4), "Ti"), "1Ti")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(1024**5), "Pi"), "1Pi")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(1024**6), "Ei"), "1Ei")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(1024**2), "Ki"), "1024Ki")
+        self.assertEqual(quantity.format_quantity(
+            Decimal((1024**3) / 2), "Gi"), "0.5Gi")
         # Decimal((1024**3)/3) are 0.3333333333333333148296162562Gi; expecting to
         # be quantized to 0.3Gi
         self.assertEqual(
@@ -693,22 +981,28 @@ def test_format_quantity(self):
             "0.3Gi")
 
         # == base 1000 ==
-        self.assertEqual(quantity.format_quantity(Decimal(0.000_000_001), "n"), "1n")
-        self.assertEqual(quantity.format_quantity(Decimal(0.000_001), "u"), "1u")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(0.000_000_001), "n"), "1n")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(0.000_001), "u"), "1u")
         self.assertEqual(quantity.format_quantity(Decimal(0.001), "m"), "1m")
         self.assertEqual(quantity.format_quantity(Decimal(1_000), "k"), "1k")
-        self.assertEqual(quantity.format_quantity(Decimal(1_000_000), "M"), "1M")
-        self.assertEqual(quantity.format_quantity(Decimal(1_000_000_000), "G"), "1G")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(1_000_000), "M"), "1M")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(1_000_000_000), "G"), "1G")
         self.assertEqual(
             quantity.format_quantity(Decimal(1_000_000_000_000), "T"), "1T"
         )
         self.assertEqual(
             quantity.format_quantity(Decimal(1_000_000_000_000_000), "P"), "1P"
         )
         self.assertEqual(
-            quantity.format_quantity(Decimal(1_000_000_000_000_000_000), "E"), "1E"
+            quantity.format_quantity(
+                Decimal(1_000_000_000_000_000_000), "E"), "1E"
         )
-        self.assertEqual(quantity.format_quantity(Decimal(1_000_000), "k"), "1000k")
+        self.assertEqual(quantity.format_quantity(
+            Decimal(1_000_000), "k"), "1000k")
         # Decimal(1_000_000/3) are 333.3333333333333139307796955k; expecting to
         # be quantized to 333k
         self.assertEqual(

kubernetes/utils/__init__.py

@@ -14,7 +14,9 @@
 
 from __future__ import absolute_import
 
-from .create_from_yaml import (FailToCreateError, create_from_dict,
-                               create_from_yaml, create_from_directory)
+from .yaml_processor import (FailToCreateError, create_from_dict,
+                             create_from_yaml, create_from_directory,
+                             FailToProcessError, process_from_dict,
+                             process_from_yaml, process_from_directory)
 from .quantity import parse_quantity
-from. duration import parse_duration
+from . duration import parse_duration

kubernetes/utils/create_from_yaml.py renamed to kubernetes/utils/yaml_processor.py

@@ -1,4 +1,4 @@
-# Copyright 2019 The Kubernetes Authors.
+# Copyright 2025 The Kubernetes Authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -24,26 +24,32 @@
 LOWER_OR_NUM_FOLLOWED_BY_UPPER_RE = re.compile("([a-z0-9])([A-Z])")
 
 
-def create_from_directory(
-    k8s_client, yaml_dir=None, verbose=False, namespace="default", apply=False, **kwargs
-):
+def process_from_directory(
+        k8s_client,
+        yaml_dir=None,
+        verbose=False,
+        namespace="default",
+        apply=False,
+        action="create",
+        **kwargs):
     """
-    Perform an action from files from a directory. Pass True for verbose to
+    Perform an action (create/delete) from files from a directory. Pass True for verbose to
     print confirmation information.
 
     Input:
     k8s_client: an ApiClient object, initialized with the client args.
     yaml_dir: string. Contains the path to directory.
-    verbose: If True, print confirmation from the create action.
+    verbose: If True, print confirmation from the action.
         Default is False.
-    namespace: string. Contains the namespace to create all
+    namespace: string. Contains the namespace to process all
         resources inside. The namespace must preexist otherwise
-        the resource creation will fail. If the API object in
+        the resource processing will fail. If the API object in
         the yaml file already contains a namespace definition
         this parameter has no effect.
-    apply: bool. If True, use server-side apply for creating resources.
+    apply: bool. If True, use server-side apply for processing resources.
+    action: string. The action to perform, either "create" or "delete".
 
-    Available parameters for creating <kind>:
+    Available parameters for processing <kind>:
     :param async_req bool
     :param bool include_uninitialized: If true, partially initialized
         resources are included in the response.
@@ -55,11 +61,11 @@ def create_from_directory(
         Valid values are: - All: all dry run stages will be processed
 
     Returns:
-        The list containing the created kubernetes API objects.
+        The list containing the processed kubernetes API objects.
 
     Raises:
-        FailToCreateError which holds list of `client.rest.ApiException`
-        instances for each object that failed to create.
+        FailToProcessError which holds list of `client.rest.ApiException`
+        instances for each object that failed to process.
     """
 
     if not yaml_dir:
@@ -80,49 +86,52 @@ def create_from_directory(
 
     for file in files:
         try:
-            k8s_objects = create_from_yaml(
+            k8s_objects = process_from_yaml(
                 k8s_client,
                 file,
                 verbose=verbose,
                 namespace=namespace,
                 apply=apply,
+                action=action,
                 **kwargs,
             )
             k8s_objects_all.append(k8s_objects)
-        except FailToCreateError as failure:
+        except FailToProcessError as failure:
             failures.extend(failure.api_exceptions)
     if failures:
-        raise FailToCreateError(failures)
+        raise FailToProcessError(failures)
     return k8s_objects_all
 
 
-def create_from_yaml(
+def process_from_yaml(
     k8s_client,
     yaml_file=None,
     yaml_objects=None,
     verbose=False,
     namespace="default",
     apply=False,
+    action="create",
     **kwargs,
 ):
     """
-    Perform an action from a yaml file. Pass True for verbose to
+    Perform an action (create/delete) from a yaml file. Pass True for verbose to
     print confirmation information.
     Input:
     yaml_file: string. Contains the path to yaml file.
     k8s_client: an ApiClient object, initialized with the client args.
     yaml_objects: List[dict]. Optional list of YAML objects; used instead
         of reading the `yaml_file`. Default is None.
-    verbose: If True, print confirmation from the create action.
+    verbose: If True, print confirmation from the action.
         Default is False.
-    namespace: string. Contains the namespace to create all
+    namespace: string. Contains the namespace to process all
         resources inside. The namespace must preexist otherwise
-        the resource creation will fail. If the API object in
+        the resource processing will fail. If the API object in
         the yaml file already contains a namespace definition
         this parameter has no effect.
-    apply: bool. If True, use server-side apply for creating resources.
+    apply: bool. If True, use server-side apply for processing resources.
+    action: string. The action to perform, either "create" or "delete".
 
-    Available parameters for creating <kind>:
+    Available parameters for processing <kind>:
     :param async_req bool
     :param bool include_uninitialized: If true, partially initialized
         resources are included in the response.
@@ -134,33 +143,34 @@ def create_from_yaml(
         Valid values are: - All: all dry run stages will be processed
 
     Returns:
-        The created kubernetes API objects.
+        The processed kubernetes API objects.
 
     Raises:
-        FailToCreateError which holds list of `client.rest.ApiException`
-        instances for each object that failed to create.
+        FailToProcessError which holds list of `client.rest.ApiException`
+        instances for each object that failed to process.
     """
 
-    def create_with(objects, apply=apply):
+    def process_with(objects, apply=apply):
         failures = []
         k8s_objects = []
         for yml_document in objects:
             if yml_document is None:
                 continue
             try:
-                created = create_from_dict(
+                processed = process_from_dict(
                     k8s_client,
                     yml_document,
                     verbose,
                     namespace=namespace,
                     apply=apply,
+                    action=action,
                     **kwargs,
                 )
-                k8s_objects.append(created)
-            except FailToCreateError as failure:
+                k8s_objects.append(processed)
+            except FailToProcessError as failure:
                 failures.extend(failure.api_exceptions)
         if failures:
-            raise FailToCreateError(failures)
+            raise FailToProcessError(failures)
         return k8s_objects
 
     class Loader(yaml.loader.SafeLoader):
@@ -170,42 +180,43 @@ class Loader(yaml.loader.SafeLoader):
 
     if yaml_objects:
         yml_document_all = yaml_objects
-        return create_with(yml_document_all)
+        return process_with(yml_document_all)
     elif yaml_file:
         with open(os.path.abspath(yaml_file)) as f:
             yml_document_all = yaml.load_all(f, Loader=Loader)
-            return create_with(yml_document_all, apply)
+            return process_with(yml_document_all, apply)
     else:
         raise ValueError(
             "One of `yaml_file` or `yaml_objects` arguments must be provided"
         )
 
 
-def create_from_dict(
-    k8s_client, data, verbose=False, namespace="default", apply=False, **kwargs
+def process_from_dict(
+    k8s_client, data, verbose=False, namespace="default", apply=False, action="create", **kwargs
 ):
     """
-    Perform an action from a dictionary containing valid kubernetes
+    Perform an action (create/delete) from a dictionary containing valid kubernetes
     API object (i.e. List, Service, etc).
 
     Input:
     k8s_client: an ApiClient object, initialized with the client args.
     data: a dictionary holding valid kubernetes objects
-    verbose: If True, print confirmation from the create action.
+    verbose: If True, print confirmation from the action.
         Default is False.
-    namespace: string. Contains the namespace to create all
+    namespace: string. Contains the namespace to process all
         resources inside. The namespace must preexist otherwise
-        the resource creation will fail. If the API object in
+        the resource processing will fail. If the API object in
         the yaml file already contains a namespace definition
         this parameter has no effect.
-    apply: bool. If True, use server-side apply for creating resources.
+    apply: bool. If True, use server-side apply for processing resources.
+    action: string. The action to perform, either "create" or "delete".
 
     Returns:
-        The created kubernetes API objects.
+        The processed kubernetes API objects.
 
     Raises:
-        FailToCreateError which holds list of `client.rest.ApiException`
-        instances for each object that failed to create.
+        FailToProcessError which holds list of `client.rest.ApiException`
+        instances for each object that failed to process.
     """
     # If it is a list type, will need to iterate its items
     api_exceptions = []
@@ -222,48 +233,58 @@ def create_from_dict(
                 yml_object["apiVersion"] = data["apiVersion"]
                 yml_object["kind"] = kind
             try:
-                created = create_from_yaml_single_item(
+                processed = process_from_yaml_single_item(
                     k8s_client,
                     yml_object,
                     verbose,
                     namespace=namespace,
                     apply=apply,
+                    action=action,
                     **kwargs,
                 )
-                k8s_objects.append(created)
+                k8s_objects.append(processed)
             except client.rest.ApiException as api_exception:
                 api_exceptions.append(api_exception)
     else:
         # This is a single object. Call the single item method
         try:
-            created = create_from_yaml_single_item(
-                k8s_client, data, verbose, namespace=namespace, apply=apply, **kwargs
-            )
-            k8s_objects.append(created)
+            processed = process_from_yaml_single_item(
+                k8s_client,
+                data,
+                verbose,
+                namespace=namespace,
+                apply=apply,
+                action=action,
+                **kwargs)
+            k8s_objects.append(processed)
         except client.rest.ApiException as api_exception:
             api_exceptions.append(api_exception)
 
     # In case we have exceptions waiting for us, raise them
     if api_exceptions:
-        raise FailToCreateError(api_exceptions)
+        raise FailToProcessError(api_exceptions)
 
     return k8s_objects
 
 
-def create_from_yaml_single_item(
-    k8s_client, yml_object, verbose=False, apply=False, **kwargs
+def process_from_yaml_single_item(
+    k8s_client, yml_object, verbose=False, apply=False, action="create", **kwargs
 ):
-
     kind = yml_object["kind"]
     if apply is True:
         apply_client = DynamicClient(k8s_client).resources.get(
             api_version=yml_object["apiVersion"], kind=kind
         )
-        resp = apply_client.server_side_apply(
-            body=yml_object, field_manager="python-client", **kwargs
-        )
+        if action == "create":
+            resp = apply_client.server_side_apply(
+                body=yml_object, field_manager="python-client", **kwargs
+            )
+        elif action == "delete":
+            resp = apply_client.delete(
+                body=yml_object, field_manager="python-client", **kwargs
+            )
         if verbose:
-            msg = "{0} created.".format(kind)
+            msg = "{0} {1}d.".format(kind, action)
             if hasattr(resp, "status"):
                 msg += " status='{0}'".format(str(resp.status))
             print(msg)
@@ -283,30 +304,41 @@ def create_from_yaml_single_item(
     # Replace CamelCased action_type into snake_case
     kind = UPPER_FOLLOWED_BY_LOWER_RE.sub(r"\1_\2", kind)
     kind = LOWER_OR_NUM_FOLLOWED_BY_UPPER_RE.sub(r"\1_\2", kind).lower()
-    # Expect the user to create namespaced objects more often
-    if hasattr(k8s_api, "create_namespaced_{0}".format(kind)):
+    # Prepare the arguments for the API call
+    api_args = {
+        "body": client.V1DeleteOptions(
+            propagation_policy="Background",
+            grace_period_seconds=5
+        ) if action == "delete" else yml_object,
+        **kwargs
+    }
+
+    if action == "delete":
+        name = yml_object["metadata"]["name"]
+        api_args["name"] = name
+
+    # Expect the user to process namespaced objects more often
+    if hasattr(k8s_api, "{0}_namespaced_{1}".format(action, kind)):
         # Decide which namespace we are going to put the object in,
         # if any
         if "namespace" in yml_object["metadata"]:
             namespace = yml_object["metadata"]["namespace"]
-            kwargs["namespace"] = namespace
-        resp = getattr(k8s_api, "create_namespaced_{0}".format(kind))(
-            body=yml_object, **kwargs
-        )
+            api_args["namespace"] = namespace
+        resp = getattr(k8s_api, "{0}_namespaced_{1}".format(
+            action, kind))(**api_args)
     else:
-        kwargs.pop("namespace", None)
-        resp = getattr(k8s_api, "create_{0}".format(kind))(
-            body=yml_object, **kwargs
-        )
+        api_args.pop("namespace", None)
+        resp = getattr(k8s_api, "{0}_{1}".format(action, kind))(**api_args)
+
     if verbose:
-        msg = "{0} created.".format(kind)
+        msg = "{0} {1}d.".format(kind, action)
         if hasattr(resp, "status"):
             msg += " status='{0}'".format(str(resp.status))
         print(msg)
     return resp
 
 
-class FailToCreateError(Exception):
+class FailToProcessError(Exception):
     """
     An exception class for handling error if an error occurred when
     handling a yaml file.
@@ -322,3 +354,137 @@ def __str__(self):
                 api_exception.reason, api_exception.body
             )
         return msg
+
+
+def create_from_directory(
+    k8s_client, yaml_dir=None, verbose=False, namespace="default", apply=False, **kwargs
+):
+    """
+    Perform an action from files from a directory. Pass True for verbose to
+    print confirmation information.
+
+    Input:
+    k8s_client: an ApiClient object, initialized with the client args.
+    yaml_dir: string. Contains the path to directory.
+    verbose: If True, print confirmation from the create action.
+        Default is False.
+    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
+        this parameter has no effect.
+    apply: bool. If True, use server-side apply for creating resources.
+
+    Available parameters for creating <kind>:
+    :param async_req bool
+    :param bool include_uninitialized: If true, partially initialized
+        resources are included in the response.
+    :param str pretty: If 'true', then the output is pretty printed.
+    :param str dry_run: When present, indicates that modifications
+        should not be persisted. An invalid or unrecognized dryRun
+        directive will result in an error response and no further
+        processing of the request.
+        Valid values are: - All: all dry run stages will be processed
+
+    Returns:
+        The list containing the created kubernetes API objects.
+
+    Raises:
+        FailToCreateError which holds list of `client.rest.ApiException`
+        instances for each object that failed to create.
+    """
+    return process_from_directory(
+        k8s_client,
+        yaml_dir,
+        verbose=verbose,
+        namespace=namespace,
+        apply=apply,
+        action="create",
+        **kwargs,
+    )
+
+
+def create_from_yaml(
+        k8s_client,
+        yaml_file=None,
+        yaml_objects=None,
+        verbose=False,
+        namespace="default",
+        apply=False,
+        **kwargs):
+    """
+    Perform an action from a yaml file. Pass True for verbose to
+    print confirmation information.
+    Input:
+    yaml_file: string. Contains the path to yaml file.
+    k8s_client: an ApiClient object, initialized with the client args.
+    yaml_objects: List[dict]. Optional list of YAML objects; used instead
+        of reading the `yaml_file`. Default is None.
+    verbose: If True, print confirmation from the create action.
+        Default is False.
+    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
+        this parameter has no effect.
+
+    Available parameters for creating <kind>:
+    :param async_req bool
+    :param bool include_uninitialized: If true, partially initialized
+        resources are included in the response.
+    :param str pretty: If 'true', then the output is pretty printed.
+    :param str dry_run: When present, indicates that modifications
+        should not be persisted. An invalid or unrecognized dryRun
+        directive will result in an error response and no further
+        processing of the request.
+        Valid values are: - All: all dry run stages will be processed
+
+    Returns:
+        The created kubernetes API objects.
+
+    Raises:
+        FailToCreateError which holds list of `client.rest.ApiException`
+        instances for each object that failed to create.
+    """
+    return process_from_yaml(
+        k8s_client,
+        yaml_file,
+        yaml_objects,
+        verbose,
+        namespace,
+        apply,
+        action="create",
+        **kwargs)
+
+
+def create_from_dict(k8s_client, data, verbose=False, namespace='default',
+                     **kwargs):
+    """
+    Perform an action from a dictionary containing valid kubernetes
+    API object (i.e. List, Service, etc).
+
+    Input:
+    k8s_client: an ApiClient object, initialized with the client args.
+    data: a dictionary holding valid kubernetes objects
+    verbose: If True, print confirmation from the create action.
+        Default is False.
+    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
+        this parameter has no effect.
+
+    Returns:
+        The created kubernetes API objects.
+
+    Raises:
+        FailToCreateError which holds list of `client.rest.ApiException`
+        instances for each object that failed to create.
+    """
+    return process_from_dict(k8s_client, data, verbose, namespace, action="create", **kwargs)
+
+class FailToCreateError(FailToProcessError):
+    """
+    An exception class for handling error if an error occurred when
+    handling a yaml file.
+    """