Automating Kubernetes · Prasanta Kumar Behera

Use the official Kubernetes Python client: connect with kubeconfig or in-cluster credentials, list and filter objects, patch resources, watch for changes, and handle API errors, with least-privilege RBAC for your tool.

Why the client library, not kubectl + subprocess?

Parsing kubectl output works for quick scripts. For real tools, the official Python client gives you typed objects, proper exceptions, watches and no dependency on the kubectl binary.

Using kubectl from Python is like asking a friend to phone the school office for you and repeat back what they said. The client library is phoning the office yourself: you hear the answer directly, in full, and you can ask follow-up questions straight away.

$ pip install kubernetes

Connect: laptop or in-cluster

from kubernetes import client, config

def connect() -> None:
    try:
        config.load_incluster_config()      # running in a pod: use its ServiceAccount
    except config.ConfigException:
        config.load_kube_config()           # on a laptop: use ~/.kube/config

connect()
v1 = client.CoreV1Api()
apps = client.AppsV1Api()

List and filter

for node in v1.list_node().items:
    ready = next(c.status for c in node.status.conditions if c.type == "Ready")
    print(f"{node.metadata.name:<20} Ready={ready:<5} kubelet={node.status.node_info.kubelet_version}")

pods = v1.list_pod_for_all_namespaces(field_selector="status.phase!=Running")
for p in pods.items:
    print(p.metadata.namespace, p.metadata.name, p.status.phase)

Objects use snake_case attributes (node_info, kubelet_version) where the YAML uses camelCase. Filter on the server with label_selector and field_selector instead of downloading everything.

Change things: patch

from kubernetes.client.rest import ApiException

def scale(name: str, namespace: str, replicas: int) -> None:
    try:
        apps.patch_namespaced_deployment_scale(name, namespace, {"spec": {"replicas": replicas}})
    except ApiException as e:
        if e.status == 404:
            raise SystemExit(f"deployment {namespace}/{name} not found")
        if e.status == 403:
            raise SystemExit(f"not allowed to scale {namespace}/{name}: check RBAC")
        raise

scale("web", "default", 3)

Cordon a node (the same as kubectl cordon):

v1.patch_node("w2", {"spec": {"unschedulable": True}})

Watch for changes

from kubernetes import watch

w = watch.Watch()
for event in w.stream(v1.list_namespaced_pod, namespace="default", timeout_seconds=120):
    pod = event["object"]
    print(event["type"], pod.metadata.name, pod.status.phase)    # ADDED / MODIFIED / DELETED

Watches are how controllers work: react to changes instead of polling (see Kubernetes Administration, lesson 26, on API load).

Least privilege for your tool

When the tool runs in the cluster, give its ServiceAccount only what it needs (see Kubernetes Administration, lesson 15):

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: node-reporter
rules:
  - apiGroups: [""]
    resources: ["nodes", "pods"]
    verbs: ["get", "list", "watch"]

Try it: a cluster report (kind cluster)

Print every node with Ready status, kubelet version and allocatable CPU/memory (node.status.allocatable).
List pods that aren't Running or Succeeded across all namespaces, with their reason (pod.status.container_statuses[*].state.waiting.reason, when present).
Scale a Deployment from 2 to 4 with your scale() function, and handle a misspelt name gracefully.
Watch the default namespace for 60 seconds while you delete a pod in another terminal.
Bonus: run the report as a Job inside the cluster with a ServiceAccount bound to the node-reporter ClusterRole.

Going deeper: beyond scripts

For long-running controllers, use frameworks that handle watches, retries and resync for you: kopf in Python, or controller-runtime/Kubebuilder in Go.
Custom resources are reached through client.CustomObjectsApi() (group, version, plural).
Use server-side apply for idempotent "make it look like this" updates instead of read-modify-write patches.
Respect API Priority and Fairness: back off on 429 responses, and never loop full LISTs.

Recap

pip install kubernetes; connect in-cluster first, kubeconfig as fallback.
CoreV1Api (pods, nodes, services…) and AppsV1Api (deployments…); filter server-side with selectors.
Patch to change, watch to react, handle ApiException by status code.
Give the tool's identity least-privilege RBAC.

This site is a public version of my personal engineering knowledge hub. It intentionally excludes confidential company information and internal operational details.