k8s python api二次封装 例子

k8s python api二次封装


pip install pprint   kubernetes
import urllib3
from pprint import pprint
from kubernetes import client
from os import path
import yaml

class K8sApi(object):
    def __init__(self):
        # self.config = config.kube_config.load_kube_config()
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        self.configuration = client.Configuration()
        self.configuration.host = "https://192.168.100.111:6443"
        self.configuration.api_key[‘authorization‘] = ‘Bearer token‘
        self.configuration.verify_ssl = False
        self.k8s_apps_v1 = client.AppsV1Api(client.ApiClient(self.configuration))
        self.Api_Instance = client.CoreV1Api(client.ApiClient(self.configuration))
        self.Api_Instance_Extensions = client.ExtensionsV1beta1Api(client.ApiClient(self.configuration))

    ####################################################################################################################

    def list_deployment(self, namespace="default"):
        api_response = self.k8s_apps_v1.list_namespaced_deployment(namespace)
        return api_response

    def read_deployment(self, name="nginx-deployment", namespace="default"):
        api_response = self.k8s_apps_v1.read_namespaced_deployment(name, namespace)
        return api_response

    def create_deployment(self, file="deploy-nginx.yaml"):
        with open(path.join(path.dirname(__file__), file)) as f:
            dep = yaml.safe_load(f)
            resp = self.k8s_apps_v1.create_namespaced_deployment(
                body=dep, namespace="default")
            return resp

    def replace_deployment(self, file="deploy-nginx.yaml", name="nginx-deployment", namespace="default"):
        with open(path.join(path.dirname(__file__), file)) as f:
            dep = yaml.safe_load(f)
            resp = self.k8s_apps_v1.replace_namespaced_deployment(name, namespace,
                                                                  body=dep)
            return resp

    def delete_deployment(self, name="nginx-deployment", namespace="default"):
        api_response = self.k8s_apps_v1.delete_namespaced_deployment(name, namespace)
        return api_response

    ####################################################################################################################

    def list_node(self):
        api_response = self.Api_Instance.list_node()
        data = {}
        for i in api_response.items:
            data[i.metadata.name] = {"name": i.metadata.name,
                                     "status": i.status.conditions[-1].type if i.status.conditions[
                                                                                   -1].status == "True" else "NotReady",
                                     "ip": i.status.addresses[0].address,
                                     "kubelet_version": i.status.node_info.kubelet_version,
                                     "os_image": i.status.node_info.os_image,
                                     }
        return data

    def list_pod(self):
        api_response = self.Api_Instance.list_pod_for_all_namespaces()
        data = {}
        for i in api_response.items:
            data[i.metadata.name] = {"ip": i.status.pod_ip, "namespace": i.metadata.namespace}
        return data

    def read_pod(self, name="nginx-pod", namespace="default"):
        api_response = self.Api_Instance.read_namespaced_pod(name, namespace)
        return api_response

    def create_pod(self, file="pod-nginx.yaml", namespace="default"):
        with open(path.join(path.dirname(__file__), file)) as f:
            dep = yaml.safe_load(f)
            api_response = self.Api_Instance.create_namespaced_pod(namespace, body=dep)
            return api_response

    def replace_pod(self, name="nginx-pod", namespace="default"):
        dep = self.read_pod(name)
        api_response = self.Api_Instance.replace_namespaced_pod(name, namespace, body=dep)
        return api_response

    def delete_pod(self, name="nginx-pod", namespace="default"):
        api_response = self.Api_Instance.delete_namespaced_pod(name, namespace)
        return api_response

    ####################################################################################################################

    def list_service(self):
        api_response = self.Api_Instance.list_service_for_all_namespaces()
        return api_response

    def read_service(self, name="", namespace="default"):
        api_response = self.Api_Instance.read_namespaced_service(name, namespace)
        return api_response

    def create_service(self, file="service-nginx.yaml", namespace="default"):
        with open(path.join(path.dirname(__file__), file)) as f:
            dep = yaml.safe_load(f)
            api_response = self.Api_Instance.create_namespaced_service(namespace, body=dep)
            return api_response

    def replace_service(self, name="hequan", namespace="default"):
        dep = self.read_pod(name)
        api_response = self.Api_Instance.replace_namespaced_service(name, namespace, body=dep)
        return api_response

    def delete_service(self, name="hequan", namespace="default"):
        api_response = self.Api_Instance.delete_namespaced_service(name, namespace)
        return api_response

    ####################################################################################################################

    def list_ingress(self):
        api_response = self.Api_Instance_Extensions.list_ingress_for_all_namespaces()
        return api_response

    def read_ingress(self, name="", namespace="default"):
        api_response = self.Api_Instance_Extensions.read_namespaced_ingress(name, namespace)
        return api_response

    def create_ingress(self, file="ingress-nginx.yaml", namespace="default"):
        with open(path.join(path.dirname(__file__), file)) as f:
            dep = yaml.safe_load(f)
            api_response = self.Api_Instance_Extensions.create_namespaced_ingress(namespace, body=dep)
            return api_response

    def replace_ingress(self, name="", file="ingress-nginx.yaml", namespace="default"):
        with open(path.join(path.dirname(__file__), file)) as f:
            dep = yaml.safe_load(f)
            api_response = self.Api_Instance_Extensions.replace_namespaced_ingress(name=name, namespace=namespace,
                                                                                   body=dep)
            return api_response

    def delete_ingress(self, name="hequan", namespace="default"):
        api_response = self.Api_Instance_Extensions.delete_namespaced_ingress(name, namespace)
        return api_response

    #####################################################################################################################

    def list_stateful(self):
        api_response = self.k8s_apps_v1.list_stateful_set_for_all_namespaces()
        return api_response

    def read_stateful(self, name="nginx-deployment", namespace="default"):
        api_response = self.k8s_apps_v1.read_namespaced_stateful_set(name, namespace)
        return api_response

    def create_stateful(self, file="deploy-nginx.yaml"):
        with open(path.join(path.dirname(__file__), file)) as f:
            dep = yaml.safe_load(f)
            resp = self.k8s_apps_v1.create_namespaced_stateful_set(
                body=dep, namespace="default")
            return resp

    def replace_stateful(self, file="deploy-nginx.yaml", name="nginx-deployment", namespace="default"):
        with open(path.join(path.dirname(__file__), file)) as f:
            dep = yaml.safe_load(f)
            resp = self.k8s_apps_v1.replace_namespaced_stateful_set(name, namespace,
                                                                    body=dep)
            return resp

    def delete_stateful(self, name="nginx-deployment", namespace="default"):
        api_response = self.k8s_apps_v1.delete_namespaced_stateful_set(name, namespace)
        return api_response

    ####################################################################################################################

if __name__ == ‘__main__‘:
    def test():
        obj = K8sApi()
        ret = obj.list_deployment()
        pprint(ret)

    test()

原文地址:https://blog.51cto.com/hequan/2438485

时间: 2024-11-07 09:13:36

k8s python api二次封装 例子的相关文章

selenium + python自动化测试unittest框架学习(五)webdriver的二次封装

因为webdriver的api方法很长,再加上大多数的定位方式是以xpath方式定位,更加让代码看起来超级长,为了使整体的代码看起来整洁,对webdriver进行封装,学习资料来源于虫师的<selenium +  python自动化测试>一书. 在与测试用例文件夹同一目录下新建一个文件夹package,用来放置封装方法的模块文件 我们将webdriver二次封装的文件命名为location.py from selenium import webdriver from test_case.pub

Selenium二次封装-Python版本

1 from selenium import webdriver 2 from selenium.webdriver.support.wait import WebDriverWait 3 from selenium.webdriver.support import expected_conditions as EC 4 from selenium.webdriver.common.by import By 5 from selenium.webdriver.common.action_chai

python+selenium十:selenium的二次封装

from selenium import webdriverfrom selenium.webdriver.support.wait import WebDriverWaitfrom selenium.webdriver.support import expected_conditions as ECfrom selenium.webdriver.support.select import Selectfrom selenium.webdriver.common.action_chains im

多线程C调用python api的陷阱

众所周知,用脚本语言编写的服务(wsgi接口)都需要一个server容器,常见的如php的php-fpm, lightd等.python中一般是用的uwsgi,uwsgi是在wsgi的基础上的一种新的协议,可以用来部署python等脚本程序的运行.然而在不熟悉uwsgi的代码架构和c调用python的api情况下进行开发可能会遇到一些意想不到的问题. 我们先看一段代码,下面这段代码是用的Flask框架,每次请求的时候会把COUNT的值先减一再加一,最后再乘二.如果请求50次,其最终的结果应该是2

Python Api接口自动化测试框架 代码写用例

公司新来两个妹子一直吐槽这个接口测试用例用excel维护起来十分费脑费事,而且比较low(内心十分赞同但是不能推翻自己),妹子说excel本来就很麻烦的工具,于是偷偷的进行了二次改版. 变更内容如下: 1.代码结构 image.png 2.新增测试报告网页版和版本管理 3.新增用例代码化 一.封装一个获取用例的模块 image.png 用例的写法可以按照yml文件的写法,后缀的文件都可为.conf..config..ini.[]中的是测试用例场景,下面的参数内容对应接口用例参数. 简单介绍下py

自动化运维工具Ansible之Python API

Ansible 的Python API使用起来相当简单快捷,使用API可以将某些运维操作封装成一个带有WEB界面的操作,免去了每次执行某个操作的时候都需要SSH运行Ansible命令. 官方给出的一个简单示例: import ansible.runner          runner = ansible.runner.Runner(        module_name='ping',        module_args='',        pattern='web*',        f

python实现人人网登录的例子

例子,python实现登录人人网.原文地址:http://www.jbxue.com/article/python/22994.html #!/usr/bin/python # www.jbxue.com import re import urllib2 import cookielib def renren(): cj = cookielib.LWPCookieJar() opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj)

【Python】一个简单的例子

问题描述: Python基础篇 参考资料: (1)http://www.cnblogs.com/octobershiner/archive/2012/12/04/2801670.html (2)http://www.cnblogs.com/itech/archive/2010/06/20/1760345.html 例子: 求解Fibonacci glb_var.py gl_count=1 path.py # coding:utf-8 ''' Created on 2014-4-28 @autho

毕加索的艺术——Picasso,一个强大的Android图片下载缓存库,OkHttpUtils的使用,二次封装PicassoUtils实现微信精选

毕加索的艺术--Picasso,一个强大的Android图片下载缓存库,OkHttpUtils的使用,二次封装PicassoUtils实现微信精选 官网: http://square.github.io/picasso/ 我们在上篇OkHttp的时候说过这个Picasso,学名毕加索,是Square公司开源的一个Android图形缓存库,而且使用起来也是非常的简单,只要一行代码就轻松搞定了,你会问,为什么不介绍一下Glide?其实Glide我有时间也是会介绍的,刚好上篇我们用到了Picasso,