kubernetes Python Client

from kubernetes import client, config

# Load from ~/.kube/config (local development)
config.load_kube_config()

# Load from in-cluster service account (when running in a pod)
# config.load_incluster_config()

# Create API clients
v1 = client.CoreV1Api()
apps_v1 = client.AppsV1Api()
batch_v1 = client.BatchV1Api()

# List pods in the default namespace
pods = v1.list_namespaced_pods(namespace="default")
for pod in pods.items:
    print(f"{pod.metadata.name}: {pod.status.phase}")

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

# List all pods across all namespaces
all_pods = v1.list_pod_for_all_namespaces()
for pod in all_pods.items:
    print(f"{pod.metadata.namespace}/{pod.metadata.name}: {pod.status.phase}")

# List pods with label selector
production_pods = v1.list_namespaced_pod(
    namespace="production",
    label_selector="env=production,app=web",
)

# List pods with field selector
running_pods = v1.list_namespaced_pod(
    namespace="default",
    field_selector="status.phase=Running",
)

# List services, deployments, configmaps
services = v1.list_namespaced_service(namespace="production")
deployments = apps_v1.list_namespaced_deployment(namespace="production")
configmaps = v1.list_namespaced_configmap(namespace="production")

print(f"Services: {len(services.items)}")
print(f"Deployments: {len(deployments.items)}")
print(f"ConfigMaps: {len(configmaps.items)}")

from kubernetes import client, config, watch

config.load_kube_config()
v1 = client.CoreV1Api()
w = watch.Watch()

# Watch events in a namespace
print("Watching events in production namespace...")
for event in w.stream(v1.list_namespaced_event, namespace="production"):
    evt = event["object"]
    print(f"{evt.reason}: {evt.involved_object.kind}/{evt.involved_object.name}")
    print(f"  Message: {evt.message}")
    print(f"  Type: {evt.type}, Count: {evt.count}")

    # Stop after 100 events
    if evt.count and evt.count > 100:
        w.stop()

from kubernetes import client, config, watch

config.load_kube_config()
v1 = client.CoreV1Api()
w = watch.Watch()

# Watch pods for crashes or restarts
for event in w.stream(v1.list_namespaced_pod, namespace="production"):
    pod = event["object"]
    name = pod.metadata.name
    phase = pod.status.phase

    if phase in ["Failed", "Unknown"]:
        print(f"ALERT: Pod {name} is {phase}")

    # Check for restart counts
    if pod.status.container_statuses:
        for cs in pod.status.container_statuses:
            if cs.restart_count > 5:
                print(f"WARNING: {name}/{cs.name} has restarted {cs.restart_count} times")

from kubernetes import client, config
import tempfile
import os

config.load_kube_config()
v1 = client.CoreV1Api()

# Exec into a pod
def exec_command(namespace: str, pod: str, container: str, command: list[str]) -> str:
    """Execute a command in a pod and return the output."""
    exec = client.api_client.ApiClient()
    api = client.CoreV1Api(api_client=exec)

    resp = api.connect_get_namespaced_pod_exec(
        name=pod,
        namespace=namespace,
        command=command,
        container=container,
        stderr=True,
        stdin=False,
        stdout=True,
        tty=False,
    )
    return resp

# Run 'df -h' in a pod
output = exec_command(
    namespace="production",
    pod="web-01-abc123",
    container="nginx",
    command=["df", "-h"],
)
print(output)

# Get environment variables
output = exec_command(
    namespace="production",
    pod="web-01-abc123",
    container="nginx",
    command=["env"],
)

from kubernetes import client, config, utils
from pathlib import Path
import yaml

config.load_kube_config()
k8s_client = client.ApiClient()

# Apply a YAML manifest
def apply_manifest(manifest_path: str) -> dict:
    """Apply a Kubernetes manifest file."""
    with open(manifest_path) as f:
        manifest = yaml.safe_load(f)

    kind = manifest["kind"]
    namespace = manifest.get("metadata", {}).get("namespace", "default")

    if kind == "Deployment":
        api = client.AppsV1Api()
        name = manifest["metadata"]["name"]
        try:
            api.read_namespaced_deployment(name, namespace)
            result = api.patch_namespaced_deployment(name, namespace, manifest)
        except client.exceptions.ApiException as e:
            if e.status == 404:
                result = api.create_namespaced_deployment(namespace, manifest)
            else:
                raise
    elif kind == "Service":
        api = client.CoreV1Api()
        name = manifest["metadata"]["name"]
        try:
            api.read_namespaced_service(name, namespace)
            result = api.patch_namespaced_service(name, namespace, manifest)
        except client.exceptions.ApiException as e:
            if e.status == 404:
                result = api.create_namespaced_service(namespace, manifest)
            else:
                raise

    return {"kind": kind, "name": manifest["metadata"]["name"]}

# Apply all manifests in a directory
for manifest in Path("k8s/").glob("*.yaml"):
    result = apply_manifest(str(manifest))
    print(f"Applied: {result['kind']}/{result['name']}")

from kubernetes import client, config
from kubernetes.client.rest import ApiException

config.load_kube_config()
v1 = client.CoreV1Api()

try:
    pod = v1.read_namespaced_pod("my-pod", "production")
    print(f"Pod status: {pod.status.phase}")
except ApiException as e:
    if e.status == 404:
        print("Pod not found")
    elif e.status == 403:
        print("Insufficient permissions")
    elif e.status == 401:
        print("Authentication failed — check kubeconfig")
    else:
        print(f"API error {e.status}: {e.reason}")
except config.ConfigException as e:
    print(f"Kubeconfig error: {e}")