云容器实例 CCI-使用kubernetes官方Python SDK访问CCI:使用Python SDK
时间:2025-02-12 14:55:06
使用Python SDK
您可以前往开发体验馆 Codelabs / Namespace生命周期代码示例(Python)下载相关代码,并在线调试。
首先需要先生成kubeconfig配置文件,参考cci-iam-authenticator使用参考,使用子命令generate-kubeconfig生成kubeconfig配置文件。

- 这里的示例代码采用了定期刷新token的方式来防止token过期(缓存值token有效期为24小时),您可以增加获取失败重试的操作,以提升可用性。
- 定期刷新token的方式不适用于该账号权限发生变更的情形,如果账号权限发生变更(如主账号变更子账号权限,导致子账号权限发生变更),变更前获取的token会失效,需要重新获取
# -*- coding: utf-8 -*-import loggingimport timeimport threadingfrom kubernetes import client, configNAMESPACE = "test-k8s-client-namespace"logging.basicConfig( level=logging.INFO, datefmt="%Y-%m-%d %H:%M:%S", format="%(asctime)s %(levelname)s %(message)s",)def create_namespace(): flavor = "general-computing" pool_size = "10" namespace = client.V1Namespace( metadata=client.V1ObjectMeta( name=NAMESPACE, annotations={ "namespace.kubernetes.io/flavor": flavor, "network.cci.io/warm-pool-size": pool_size, }, labels={ "rbac.authorization.cci.io/enable-k8s-rbac": "false", } ) ) logging.info("start to create namespace %s", NAMESPACE) client.CoreV1Api().create_namespace(namespace) logging.info("namespace created")def create_network(): name = "test-k8s-client-namespace-cn-north-7-default-network" project_id = "{project_id}" domain_id = "{domain_id}" security_group_id = "{security_group_id}" available_zone = "{available_zone}" vpc_id = "{vpc_id}" cidr = "{cidr}" network_id = "{network_id}" subnet_id = "{subnet_id}" body = { "apiVersion": "networking.cci.io/v1beta1", "kind": "Network", "metadata": { "annotations": { "network.alpha.kubernetes.io/default-security-group": security_group_id, "network.alpha.kubernetes.io/domain-id": domain_id, "network.alpha.kubernetes.io/project-id": project_id, }, "name": name, }, "spec": { "availableZone": available_zone, "cidr": cidr, "attachedVPC": vpc_id, "networkID": network_id, "networkType": "underlay_neutron", "subnetID": subnet_id, } } api = client.CustomObjectsApi() logging.info("start to create network") api.create_namespaced_custom_object( group="networking.cci.io", version="v1beta1", namespace=NAMESPACE, plural="networks", body=body, ) logging.info("network created")def create_deployment(): app = "test-k8s-client-deployment" image = "library/nginx:stable-alpine-perl" body = client.V1Deployment( api_version="apps/v1", kind="Deployment", metadata=client.V1ObjectMeta(name=app), spec=client.V1DeploymentSpec( replicas=2, selector={"matchLabels": {"app": app}}, template=client.V1PodTemplateSpec( metadata=client.V1ObjectMeta(labels={"app": app}), spec=client.V1PodSpec( containers=[ client.V1Container( name="container-0", image=image, resources=client.V1ResourceRequirements( requests={"cpu": "500m", "memory": "1024Mi"}, limits={"cpu": "500m", "memory": "1024Mi"}, ), ) ], image_pull_secrets=[ client.V1LocalObjectReference(name="imagepull-secret")], priority=0), ), ) ) logging.info("start to create deployment %s/%s", NAMESPACE, app) client.AppsV1Api().create_namespaced_deployment(NAMESPACE, body) logging.info("deployment created")def get_deployment(): app = "test-k8s-client-deployment" resp = client.AppsV1Api().read_namespaced_deployment(app, NAMESPACE) logging.info("deployment detail: %s", resp)def delete_deployment(): app = "test-k8s-client-deployment" logging.info("start to delete deployment") client.AppsV1Api().delete_namespaced_deployment(app, NAMESPACE) logging.info("deployment deleted")def delete_namespace(): logging.info("start to delete namespace: %s", NAMESPACE) client.CoreV1Api().delete_namespace(NAMESPACE)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. path = '{path to kubeconfig}' config.load_kube_config(path) # 因为token有效期为24小时,所以这里设置了一个每12小时获取新的token的定时任务 # 注意:如果账号权限发生变更(如主账号变更子账号权限,导致子账号权限发生变更),变更前获取的token会失效,需要重新获取。 # 另外,您可以增加获取失败重试的操作,以提升可用性 def _refresh(): while True: time.sleep(12 * 3600) try: config.load_kube_config(path) except Exception as e: print("load_kube_config error: %s" % e) t = threading.Thread(target=_refresh) t.daemon = True t.start() create_namespace() create_network() # wait for namespace and network to be active logging.info("waiting for namespace and network to be active") time.sleep(30) create_deployment() get_deployment() delete_deployment() delete_namespace()if __name__ == '__main__': main()
support.huaweicloud.com/sdkreference-cci/cci_09_0003.html
看了此文的人还看了
CDN加速
GaussDB
文字转换成语音
免费的服务器
如何创建网站
域名网站购买
私有云桌面
云主机哪个好
域名怎么备案
手机云电脑
SSL证书申请
云点播服务器
免费OCR是什么
电脑云桌面
域名备案怎么弄
语音转文字
文字图片识别
云桌面是什么
网址安全检测
网站建设搭建
国外CDN加速
SSL免费证书申请
短信批量发送
图片OCR识别
云数据库MySQL
个人域名购买
录音转文字
扫描图片识别文字
OCR图片识别
行驶证识别
虚拟电话号码
电话呼叫中心软件
怎么制作一个网站
Email注册网站
华为VNC
图像文字识别
企业网站制作
个人网站搭建
华为云计算
免费租用云托管
云桌面云服务器
ocr文字识别免费版
HTTPS证书申请
图片文字识别转换
国外域名注册商
使用免费虚拟主机
云电脑主机多少钱
鲲鹏云手机
短信验证码平台
OCR图片文字识别
SSL证书是什么
申请企业邮箱步骤
免费的企业用邮箱
云免流搭建教程
域名价格
推荐文章
ServerLess102