Automating Kubernetes Operations with the Python Client

This guide builds on a previous article about using Python to automate Kubernetes tasks and shows concrete examples of common operations performed via the Python client library.

To restart a deployment, the script updates the kubectl.kubernetes.io/restartedAt annotation with the current UTC timestamp and patches the deployment:

from kubernetes import dynamic
from kubernetes.client import api_client
import datetime

client = dynamic.DynamicClient(api_client.ApiClient(configuration=configuration))
api = client.resources.get(api_version="apps/v1", kind="Deployment")

deployment_manifest["spec"]["template"]["metadata"]["annotations"] = {
    "kubectl.kubernetes.io/restartedAt": datetime.datetime.utcnow().isoformat()
}

deployment_patched = api.patch(body=deployment_manifest, name=deployment_name, namespace="default")

Scaling a deployment can be done with the patch_namespaced_deployment_scale method:

from kubernetes import client
api_client = client.ApiClient(configuration)
apps_v1 = client.AppsV1Api(api_client)
api_response = apps_v1.patch_namespaced_deployment_scale(
    deployment_name, "default", {"spec": {"replicas": 5}}
)

For troubleshooting, you can exec into a pod to inspect environment variables or run arbitrary shell commands:

from kubernetes.stream import stream

def pod_exec(name, namespace, command, api_instance):
    exec_command = ["/bin/sh", "-c", command]
    resp = stream(
        api_instance.connect_get_namespaced_pod_exec,
        name,
        namespace,
        command=exec_command,
        stderr=True,
        stdin=False,
        stdout=True,
        tty=False,
        _preload_content=False,
    )
    while resp.is_open():
        resp.update(timeout=1)
        if resp.peek_stdout():
            print(f"STDOUT: \n{resp.read_stdout()}")
        if resp.peek_stderr():
            print(f"STDERR: \n{resp.read_stderr()}")
    resp.close()
    if resp.returncode != 0:
        raise Exception("Script failed")

pod = "example"
api_client = client.ApiClient(configuration)
v1 = client.CoreV1Api(api_client)
pod_exec(pod, "default", "env", v1)

Applying a node taint (which lacks a direct kubectl API) can be achieved by patching the node object:

from kubernetes import client
api_client = client.ApiClient(configuration)
v1 = client.CoreV1Api(api_client)

v1.patch_node(
    "api-playground-worker",
    {"spec": {"taints": [{"effect": "NoSchedule", "key": "some-taint", "value": "1"}]}}
)

To monitor cluster resource utilization, you can query the metrics‑server via the CustomObjectsApi and print CPU and memory usage for each node:

from kubernetes import client
api_client = client.ApiClient(configuration)
custom_api = client.CustomObjectsApi(api_client)
response = custom_api.list_cluster_custom_object(
    "metrics.k8s.io", "v1beta1", "nodes"
)
for node in response["items"]:
    print(f"{node['metadata']['name']:<30} CPU: {node['usage']['cpu']:<10} Memory: {node['usage']['memory']}")

Finally, the article shows how to convert between YAML/JSON files and Kubernetes objects using the client’s serialization utilities and the kopf library, as well as how to create resources from YAML with kubernetes.utils.create_from_yaml .

For a complete list of functions and additional examples, refer to the Python client repository’s example directory.