Simply calling mechanism

Author: micw523

examples/create_deployment_from_yaml.py

@@ -14,15 +14,16 @@
 
 from os import path
 
-from kubernetes import config, utils
+from kubernetes import client, config, utils
 
 
 def main():
     # Configs can be set in Configuration class directly or using helper
     # utility. If no argument provided, the config will be loaded from
     # default location.
     config.load_kube_config()
-    k8s_api = utils.create_from_yaml("nginx-deployment.yaml")
+    k8s_client = client.ApiClient()
+    k8s_api = utils.create_from_yaml(k8s_client, "nginx-deployment.yaml")
     deps = k8s_api.read_namespaced_deployment("nginx-deployment", "default")
     print("Deployment {0} created".format(deps.metadata.name))
 

kubernetes/e2e_test/test_utils.py

@@ -24,96 +24,93 @@ def setUpClass(cls):
         cls.config = base.get_e2e_configuration()
 
     def test_app_yaml(self):
-        k8s_api = utils.create_from_yaml(
-            "kubernetes/e2e_test/test_yaml/apps-deployment.yaml", 
-            configuration=self.config)
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        k8s_api = utils.create_from_yaml(k8s_client, 
+            "kubernetes/e2e_test/test_yaml/apps-deployment.yaml")
         self.assertEqual("apps/v1beta1", 
             k8s_api.get_api_resources().group_version)
         dep = k8s_api.read_namespaced_deployment(name="nginx-app",
             namespace="default")
-        self.assertEqual("nginx-app", dep.metadata.name)
+        self.assertIsNotNone(dep)
         resp = k8s_api.delete_namespaced_deployment(
             name="nginx-app", namespace="default", 
             body={})
 
     def test_extension_yaml(self):
-        k8s_api = utils.create_from_yaml(
-            "kubernetes/e2e_test/test_yaml/extensions-deployment.yaml",
-            configuration=self.config)
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        k8s_api = utils.create_from_yaml(k8s_client, 
+            "kubernetes/e2e_test/test_yaml/extensions-deployment.yaml")
         self.assertEqual("extensions/v1beta1", 
             k8s_api.get_api_resources().group_version)
         dep = k8s_api.read_namespaced_deployment(name="nginx-deployment", 
             namespace="default")
-        self.assertEqual("nginx-deployment", dep.metadata.name)
+        self.assertIsNotNone(dep)
         resp = k8s_api.delete_namespaced_deployment(
             name="nginx-deployment", namespace="default", 
             body={})
 
     def test_core_pod_yaml(self):
-        k8s_api = utils.create_from_yaml(
-            "kubernetes/e2e_test/test_yaml/core-pod.yaml",
-            configuration=self.config)
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        k8s_api = utils.create_from_yaml(k8s_client, 
+            "kubernetes/e2e_test/test_yaml/core-pod.yaml")
         self.assertEqual("v1", 
             k8s_api.get_api_resources().group_version)
         pod = k8s_api.read_namespaced_pod(name="myapp-pod", 
             namespace="default")
-        self.assertEqual("myapp-pod", pod.metadata.name)
+        self.assertIsNotNone(pod)
         resp = k8s_api.delete_namespaced_pod(
             name="myapp-pod", namespace="default",
             body={})
 
     def test_core_service_yaml(self):
-        k8s_api = utils.create_from_yaml(
-            "kubernetes/e2e_test/test_yaml/core-service.yaml",
-            configuration=self.config)
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        k8s_api = utils.create_from_yaml(k8s_client, 
+            "kubernetes/e2e_test/test_yaml/core-service.yaml")
         self.assertEqual("v1", 
             k8s_api.get_api_resources().group_version)
         svc = k8s_api.read_namespaced_service(name="my-service",
             namespace="default")
-        self.assertEqual("my-service", svc.metadata.name)
+        self.assertIsNotNone(svc)
         resp = k8s_api.delete_namespaced_service(
             name="my-service", namespace="default",
             body={})
 
     def test_core_namespace_yaml(self):
-        k8s_api = utils.create_from_yaml(
-            "kubernetes/e2e_test/test_yaml/core-namespace.yaml",
-            configuration=self.config)
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        k8s_api = utils.create_from_yaml(k8s_client, 
+            "kubernetes/e2e_test/test_yaml/core-namespace.yaml")
         self.assertEqual("v1", 
             k8s_api.get_api_resources().group_version)
         nmsp = k8s_api.read_namespace(name="development")
-        self.assertEqual("development", nmsp.metadata.name)
+        self.assertIsNotNone(nmsp)
         resp = k8s_api.delete_namespace(name="development", body={})
 
     def test_deployment_in_namespace(self):
         k8s_client = client.ApiClient(configuration=self.config)
-        core_api = client.CoreV1Api(k8s_client)
-        core_api = utils.create_from_yaml(
-            "kubernetes/e2e_test/test_yaml/core-namespace-dep.yaml",
-            configuration=self.config)
+        core_api = utils.create_from_yaml(k8s_client, 
+            "kubernetes/e2e_test/test_yaml/core-namespace-dep.yaml")
         self.assertEqual("v1", 
             core_api.get_api_resources().group_version)
         nmsp = core_api.read_namespace(name="dep")
-        self.assertEqual("dep", nmsp.metadata.name)
-        dep_api = utils.create_from_yaml(
-            "kubernetes/e2e_test/test_yaml/extensions-deployment-dep.yaml",
-            configuration=self.config)
+        self.assertIsNotNone(nmsp)
+        dep_api = utils.create_from_yaml(k8s_client, 
+            "kubernetes/e2e_test/test_yaml/extensions-deployment-dep.yaml")
         dep = dep_api.read_namespaced_deployment(name="nginx-deployment", 
             namespace="dep")
-        self.assertEqual("nginx-deployment", dep.metadata.name)
+        self.assertIsNotNone(dep)
         resp = dep_api.delete_namespaced_deployment(
             name="nginx-deployment", namespace="dep", 
             body={})
         resp = core_api.delete_namespace(name="dep", body={})
 
     def test_api_service(self):
-        k8s_api = utils.create_from_yaml(
-            "kubernetes/e2e_test/test_yaml/api-service.yaml",
-            configuration=self.config)
+        k8s_client = client.api_client.ApiClient(configuration=self.config)
+        k8s_api = utils.create_from_yaml(k8s_client, 
+            "kubernetes/e2e_test/test_yaml/api-service.yaml")
         self.assertEqual("apiregistration.k8s.io/v1beta1", 
             k8s_api.get_api_resources().group_version)
         svc = k8s_api.read_api_service(
             name="v1alpha1.wardle.k8s.io")
-        self.assertEqual("v1alpha1.wardle.k8s.io", svc.metadata.name)
+        self.assertIsNotNone(svc)
         resp = k8s_api.delete_api_service(
             name="v1alpha1.wardle.k8s.io", body={})

kubernetes/utils/create_from_yaml.py

@@ -22,39 +22,25 @@
 
 from kubernetes import client
 
-def create_from_yaml(yaml_file, verbose=False, **kwargs):
+def create_from_yaml(k8s_client, yaml_file, verbose=False, **kwargs):
     """
     Perform an action from a yaml file. Pass 1 for verbose to
     print confirmation information.
-
-    Available parameters for generating the client:
-    :param configuration: Configuration for the client.
-    :param host: The base path for the server to call.
-    :param header_name: a header to pass when making calls to the API.
-    :param header_value: a header value to pass when making calls to the API.
+    
+    Input:
+    yaml_file: string. Contains the path to yaml file.
+    k8s_cline: an ApiClient object, initialized with the client args.
 
     Available parameters for performing the subsequent action:
     :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
     """
-    
-    client_params = ["configuration", "host", "header_name", "header_value"]
-    action_params = ["async_req", "include_unitialized", "pretty", "dry_run",
-        '_return_http_data_only', '_preload_content', '_request_timeout']
-    client_args = {}
-    action_args = {}
-    params = locals()
-    for key, val in iteritems(params["kwargs"]):
-        if key in client_params:
-            client_args[key] = val
-        elif key in action_params:
-            action_args[key] = val
 
-    k8s_client = client.api_client.ApiClient(**client_args)
     with open(path.abspath(yaml_file)) as f:
         yml_object = yaml.load(f)
+        #TODO: case of yaml file containing multiple objects
         api_type, _, api_version = yml_object["apiVersion"].partition("/")
         if api_version == "":
             api_version = api_type
@@ -78,10 +64,10 @@ def create_from_yaml(yaml_file, verbose=False, **kwargs):
         # Expect the user to create namespaced objects more often
         if hasattr(k8s_api, "create_namespaced_{0}".format(action_type_camel)):
             resp = getattr(k8s_api, "create_namespaced_{0}".format(action_type_camel))(
-                body=yml_object, namespace=dep_namespace, **action_args)
+                body=yml_object, namespace=dep_namespace, **kwargs)
         else:
             resp = getattr(k8s_api, "create_{0}".format(action_type_camel))(
-                body=yml_object, **action_args)
+                body=yml_object, **kwargs)
         if verbose:
             print("{0} created. status='{1}'".format(action_type, str(resp.status)))
         return k8s_api