tutorials/ApacheCNF/automation/k8s_client.py

   1 # ============LICENSE_START=======================================================
   2 # Copyright (C) 2021 Orange
   3 # ================================================================================
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 #
  16 # ============LICENSE_END=========================================================
  17
  18 import os
  19 from pprint import pprint
  20
  21 import oyaml as yaml
  22 from kubernetes import config, client
  23 from kubernetes.client import OpenApiException
  24
  25
  26 class K8sClient:
  27     def __init__(self, kubeconfig_path):
  28         self.mypath = os.path.dirname(os.path.realpath(__file__))
  29         config.load_kube_config(config_file=os.path.join(self.mypath, kubeconfig_path))
  30         self.api_instance = client.CustomObjectsApi()
  31
  32     def read_custom_object_file(self, file_path):
  33         with open(file_path) as crd_file:
  34             crd_body = yaml.load(crd_file, Loader=yaml.FullLoader)
  35         return crd_body
  36
  37     def get_custom_object_details(self, crd_body):
  38         group = crd_body["apiVersion"].split("/")[0]
  39         version = crd_body["apiVersion"].split("/")[1]
  40         plural = crd_body["kind"].lower() + "s"
  41         #name = crd_body["metadata"]["name"]
  42
  43         return group, version, plural #, name
  44
  45     def create_custom_object(self, file_path):
  46         crd_body = self.read_custom_object_file(file_path)
  47         #group, version, plural, name = self.get_custom_object_details(crd_body)
  48         group, version, plural = self.get_custom_object_details(crd_body)
  49         api_response = None
  50         try:
  51             api_response = self.api_instance.create_cluster_custom_object(group=group,
  52                                                                           version=version,
  53                                                                           plural=plural,
  54                                                                           body=crd_body,
  55                                                                           pretty="true")
  56         except OpenApiException as error:
  57             print(str(error.status) + " " + error.reason)
  58             pprint(error.body)
  59         return api_response