Python操作K8s常用查询API汇总

以下API均基于官方kubernetes客户端库，使用前需先初始化对应API实例（如CoreV1Api、AppsV1Api），基础初始化代码：

from kubernetes import client, config# 加载配置（本地/集群内）config.load_kube_config()  # 本地环境# config.load_incluster_config()  # 集群内Pod环境# 初始化核心API实例（Pod/Service/Namespace等）v1_core = client.CoreV1Api()# 初始化Apps API实例（Deployment/StatefulSet/DaemonSet等）v1_apps = client.AppsV1Api()# 初始化Networking API实例（Ingress）v1_networking = client.NetworkingV1Api()# 初始化Batch API实例（Job/CronJob）v1_batch = client.BatchV1Api()# 初始化自定义资源API实例（CRD）v1_custom = client.CustomObjectsApi()

一、核心资源查询API（CoreV1Api）

list_namespaced_pod 查询指定命名空间下所有Podnamespace：命名空间label_selector：标签筛选field_selector：字段筛选 |

示例

# 查询default命名空间下所有running状态的Podpods = v1_core.list_namespaced_pod(    namespace="default",    field_selector="status.phase=Running")for pod in pods.items:print(pod.metadata.name)

read_namespaced_pod 查询单个Pod的详细信息name：Pod名称namespace：命名空间

pod = v1_core.read_namespaced_pod(   name="nginx-78f89f979f-2x78z",   namespace="default")print(f"Pod IP: {pod.status.pod_ip}, 状态: {pod.status.phase}")

list_namespaced_service 查询指定命名空间下所有Service

namespace：命名空间label_selector：标签筛选

services = v1_core.list_namespaced_service(namespace="default")for svc in services.items:print(f"Service名称: {svc.metadata.name}, 类型: {svc.spec.type}")

read_namespaced_service 查询单个Service的详细信息name：Service名称namespace：命名空间

svc = v1_core.read_namespaced_service(   name="nginx-svc",   namespace="default")print(f"ClusterIP: {svc.spec.cluster_ip}, 端口: {svc.spec.ports[0].port}")

list_namespace 查询集群所有Namespacefield_selector：字段筛选（可选）

namespaces = v1_core.list_namespace()for ns in namespaces.items:print(ns.metadata.name)

read_namespace 查询单个Namespace的详细信息name：Namespace名称

ns = v1_core.read_namespace(name="default")print(f"Namespace创建时间: {ns.metadata.creation_timestamp}")

list_node 查询集群所有节点 label_selector：标签筛选field_selector：字段筛选

nodes = v1_core.list_node()for node in nodes.items:print(f"节点名称: {node.metadata.name}, 状态: {node.status.conditions[-1].status}")

read_node | 查询单个节点的详细信息 name：节点名称

node = v1_core.read_node(name="k8s-node-01")print(f"节点CPU数量: {node.status.capacity['cpu']}")

list_namespaced_config_map 查询指定命名空间下所有ConfigMapnamespace：命名空间

cm_list = v1_core.list_namespaced_config_map(namespace="default")for cm in cm_list.items:print(f"ConfigMap名称: {cm.metadata.name}, 数据: {cm.data}")

list_namespaced_secret 查询指定命名空间下所有Secretnamespace：命名空间

secret_list = v1_core.list_namespaced_secret(namespace="default")for secret in secret_list.items:print(f"Secret名称: {secret.metadata.name}, 类型: {secret.type}")

二、应用资源查询API（AppsV1Api）

list_namespaced_deployment 查询指定命名空间下所有Deploymentnamespace：命名空间label_selector：标签筛选

deployments = v1_apps.list_namespaced_deployment(namespace="default")for deploy in deployments.items:print(f"Deployment名称: {deploy.metadata.name}, 副本数: {deploy.spec.replicas}")

read_namespaced_deployment 查询单个Deployment的详细信息

name：Deployment名称namespace：命名空间

deploy = v1_apps.read_namespaced_deployment(    name="nginx-deploy",    namespace="default")print(f"镜像: {deploy.spec.template.spec.containers[0].image}, 就绪数: {deploy.status.ready_replicas}")

list_namespaced_stateful_set 查询指定命名空间下所有StatefulSet

namespace：命名空间

sts_list = v1_apps.list_namespaced_stateful_set(namespace="default")for sts in sts_list.items:print(f"StatefulSet名称: {sts.metadata.name}")

list_namespaced_daemon_set 查询指定命名空间下所有DaemonSetnamespace：命名空间

ds_list = v1_apps.list_namespaced_daemon_set(namespace="kube-system")for ds in ds_list.items:print(f"DaemonSet名称: {ds.metadata.name}")

三、网络资源查询API（NetworkingV1Api）

list_namespaced_ingress 查询指定命名空间下所有Ingressnamespace：命名空间

ingress_list = v1_networking.list_namespaced_ingress(namespace="default")for ingress in ingress_list.items:print(f"Ingress名称: {ingress.metadata.name}, 域名: {ingress.spec.rules[0].host}")

read_namespaced_ingress 查询单个Ingress的详细信息name：Ingress名称namespace：命名空间

ingress = v1_networking.read_namespaced_ingress(    name="nginx-ingress",    namespace="default")print(f"路径规则: {ingress.spec.rules[0].http.paths[0].path}")

四、批量任务查询API（BatchV1Api）

list_namespaced_job 查询指定命名空间下所有Jobnamespace：命名空间

job_list = v1_batch.list_namespaced_job(namespace="default")for job in job_list.items:print(f"Job名称: {job.metadata.name}, 完成数: {job.status.succeeded}")

list_namespaced_cron_job 查询指定命名空间下所有CronJobnamespace：命名空间

cronjob_list = v1_batch.list_namespaced_cron_job(namespace="default")for cj in cronjob_list.items:print(f"CronJob名称: {cj.metadata.name}, 调度规则: {cj.spec.schedule}")

五、自定义资源（CRD）查询API（CustomObjectsApi）

list_namespaced_custom_object 查询指定命名空间下自定义资源

group：CRD组名version：CRD版本namespace：命名空间plural：CRD复数名称

# 示例：查询Prometheus Operator的Prometheus CRDprom_list = v1_custom.list_namespaced_custom_object(   group="monitoring.coreos.com",   version="v1",   namespace="monitoring",   plural="prometheuses")for prom in prom_list["items"]:print(f"Prometheus名称: {prom['metadata']['name']}")

关键查询技巧补充

1. 筛选优化
   ：使用label_selector（标签筛选）和field_selector（字段筛选）减少返回数据量，提升查询效率，例如：

   # 筛选标签为app=nginx的Podpods = v1_core.list_namespaced_pod(    namespace="default",    label_selector="app=nginx")

2. 分页查询
   ：通过limit和continue_token实现大数量资源分页查询，避免单次返回数据过多：

   # 分页查询，每次返回10个Podfirst_page = v1_core.list_namespaced_pod(    namespace="default",    limit=10)# 获取下一页（如果有）if first_page.metadata._continue:    second_page = v1_core.list_namespaced_pod(        namespace="default",        limit=10,        _continue=first_page.metadata._continue    )

3. 异常处理
   ：查询时捕获ApiException，处理404（资源不存在）、403（权限不足）等异常：

   from kubernetes.client.exceptions import ApiExceptiontry:    pod = v1_core.read_namespaced_pod(name="non-exist-pod", namespace="default")except ApiException as e:if e.status == 404:print("Pod不存在")elif e.status == 403:print("权限不足")else:print(f"查询失败: {e.reason}")

最后

1. Python查询K8s的API按资源类型分布在不同的API类中（CoreV1Api/AppsV1Api等），核心是list_*（批量查询）和read_*（单个查询）两类方法；
2. namespace
   是绝大多数查询API的必选参数（集群级资源如Node/Namespace除外）；
3. 筛选参数（label_selector/field_selector）能大幅提升查询效率，是生产环境必备；
4. CRD查询需明确CRD的group/version/plural三个核心参数，需提前从CRD定义中获取。