kubernetes-python

Installation

```bash

pip install kubernetes

```

Basic Usage

```python

from kubernetes import client, config

# Load kubeconfig

config.load_kube_config()

# Create API client

v1 = client.CoreV1Api()

# List pods

pods = v1.list_pod_for_all_namespaces(limit=10)

for pod in pods.items:

print(f"{pod.metadata.namespace}/{pod.metadata.name}")

```

Client Initialization Patterns

Local Development (using kubeconfig):

```python

from kubernetes import client, config

config.load_kube_config()

v1 = client.CoreV1Api()

```

In-Cluster (running inside Kubernetes):

```python

from kubernetes import client, config

config.load_incluster_config()

v1 = client.CoreV1Api()

```

Specific Context:

```python

config.load_kube_config(context='production-cluster')

v1 = client.CoreV1Api()

```

API Clients by Resource Type

| API Client | Resources | Usage |

|-----------|-----------|-------|

| CoreV1Api | Pods, Services, ConfigMaps, Secrets, Namespaces, PVCs | client.CoreV1Api() |

| AppsV1Api | Deployments, StatefulSets, DaemonSets, ReplicaSets | client.AppsV1Api() |

| BatchV1Api | Jobs, CronJobs | client.BatchV1Api() |

| NetworkingV1Api | Ingresses, NetworkPolicies | client.NetworkingV1Api() |

| CustomObjectsApi | Custom Resources (CRDs) | client.CustomObjectsApi() |

Creating a Deployment

```python

from kubernetes import client

apps_v1 = client.AppsV1Api()

deployment = client.V1Deployment(

metadata=client.V1ObjectMeta(name="nginx-deployment"),

spec=client.V1DeploymentSpec(

replicas=3,

selector=client.V1LabelSelector(

match_labels={"app": "nginx"}

),

template=client.V1PodTemplateSpec(

metadata=client.V1ObjectMeta(labels={"app": "nginx"}),

spec=client.V1PodSpec(

containers=[

client.V1Container(

name="nginx",

image="nginx:1.14.2",

ports=[client.V1ContainerPort(container_port=80)]

)

]

)

)

)

)

apps_v1.create_namespaced_deployment(

namespace="default",

body=deployment

)

```

Reading Resources

```python

# Read single pod

pod = v1.read_namespaced_pod(name="my-pod", namespace="default")

# List pods with label selector

pods = v1.list_namespaced_pod(

namespace="default",

label_selector="app=nginx,env=production"

)

# List pods with field selector

running_pods = v1.list_namespaced_pod(

namespace="default",

field_selector="status.phase=Running"

)

```

Updating Resources

Patch (partial update, preferred):

```python

deployment = apps_v1.read_namespaced_deployment(

name="nginx-deployment",

namespace="default"

)

deployment.spec.replicas = 5

apps_v1.patch_namespaced_deployment(

name="nginx-deployment",

namespace="default",

body=deployment

)

```

Replace (full update):

```python

deployment.metadata.resource_version = existing.metadata.resource_version

apps_v1.replace_namespaced_deployment(

name="nginx-deployment",

namespace="default",

body=deployment

)

```

Deleting Resources

```python

v1.delete_namespaced_pod(

name="my-pod",

namespace="default"

)

# Delete with grace period

apps_v1.delete_namespaced_deployment(

name="nginx-deployment",

namespace="default",

grace_period_seconds=30

)

```

```python

from kubernetes.client.rest import ApiException

try:

pod = v1.read_namespaced_pod(name="my-pod", namespace="default")

except ApiException as e:

if e.status == 404:

print("Pod not found")

elif e.status == 403:

print("Permission denied")

else:

print(f"API error: {e}")

```

Create or Update Pattern (Idempotent)

```python

from kubernetes.client.rest import ApiException

def create_or_update_deployment(apps_v1, namespace, deployment):

"""Create deployment if it doesn't exist, otherwise update it."""

name = deployment.metadata.name

try:

existing = apps_v1.read_namespaced_deployment(

name=name,

namespace=namespace

)

deployment.metadata.resource_version = existing.metadata.resource_version

response = apps_v1.replace_namespaced_deployment(

name=name,

namespace=namespace,

body=deployment

)

print(f"Deployment {name} updated")

return response

except ApiException as e:

if e.status == 404:

response = apps_v1.create_namespaced_deployment(

namespace=namespace,

body=deployment

)

print(f"Deployment {name} created")

return response

else:

raise

```

Watch for resource changes in real-time:

```python

from kubernetes import watch

w = watch.Watch()

# Watch pods with timeout

for event in w.stream(

v1.list_namespaced_pod,

namespace="default",

timeout_seconds=60

):

print(f"{event['type']}: {event['object'].metadata.name}")

w.stop()

```

Event Types

• ADDED: Resource was created
• MODIFIED: Resource was updated
• DELETED: Resource was deleted
• ERROR: Watch error occurred

ConfigMap

```python

configmap = client.V1ConfigMap(

metadata=client.V1ObjectMeta(name="my-config"),

data={"key1": "value1", "key2": "value2"}

)

v1.create_namespaced_config_map(

namespace="default",

body=configmap

)

```

Secret

```python

import base64

secret = client.V1Secret(

metadata=client.V1ObjectMeta(name="my-secret"),

type="Opaque",

data={

"username": base64.b64encode(b"admin").decode('utf-8'),

"password": base64.b64encode(b"secretpass").decode('utf-8')

}

)

v1.create_namespaced_secret(

namespace="default",

body=secret

)

```

```python

custom_api = client.CustomObjectsApi()

# List custom resources

custom_objects = custom_api.list_namespaced_custom_object(

group="example.com",

version="v1",

namespace="default",

plural="mycustomresources"

)

# Create custom resource

custom_object = {

"apiVersion": "example.com/v1",

"kind": "MyCustomResource",

"metadata": {"name": "my-cr"},

"spec": {"replicas": 3}

}

custom_api.create_namespaced_custom_object(

group="example.com",

version="v1",

namespace="default",

plural="mycustomresources",

body=custom_object

)

```

Timeout Configuration

```python

# Set timeout for operations

pods = v1.list_namespaced_pod(

namespace="default",

_request_timeout=10 # 10 second timeout

)

```

Pagination for Large Lists

```python

def list_all_pods_paginated(namespace, page_size=100):

"""List all pods with pagination."""

all_pods = []

continue_token = None

while True:

if continue_token:

response = v1.list_namespaced_pod(

namespace=namespace,

limit=page_size,

_continue=continue_token

)

else:

response = v1.list_namespaced_pod(

namespace=namespace,

limit=page_size

)

all_pods.extend(response.items)

continue_token = response.metadata._continue

if not continue_token:

break

return all_pods

```

Server-Side Filtering

```python

# Good: Filter server-side (efficient)

running_pods = v1.list_pod_for_all_namespaces(

field_selector='status.phase=Running'

)

# Bad: Fetch all and filter client-side (inefficient)

all_pods = v1.list_pod_for_all_namespaces()

running_pods = [p for p in all_pods.items if p.status.phase == 'Running']

```

For detailed information, see:

• [Client Setup](references/client-setup.md): Authentication methods, EKS integration, configuration
• [Resource Operations](references/resource-operations.md): Comprehensive CRUD patterns for all resource types
• [Watch Patterns](references/watch-patterns.md): Advanced watch patterns, resilient loops, async operations

Strengths

• Official Kubernetes client (SIG API Machinery)
• Complete API coverage (all resources)
• Production-ready and battle-tested
• Multiple authentication methods
• Real-time watch/stream capabilities
• Full CRD support

Considerations

• Auto-generated code (less Pythonic)
• Performance can degrade in very large clusters (3000+ resources)
• No native async support (use kubernetes_asyncio package)
• EKS requires manual token refresh (15-minute expiry)

Match client version to Kubernetes cluster version:

| Client Version | K8s 1.29 | K8s 1.30 | K8s 1.31 |

|---------------|----------|----------|----------|

| 29.y.z | ✓ | +- | - |

| 30.y.z | +- | ✓ | +- |

| 31.y.z | +- | +- | ✓ |

• ✓ = Exact feature/API parity
• +- = Most APIs work, some new/removed
• \- = Not recommended

1. Least Privilege: Never use cluster-admin for applications
2. RBAC: Use Roles (namespaced) instead of ClusterRoles when possible
3. Secrets: Don't log secret data, load credentials from environment
4. SSL Verification: Always verify SSL in production (verify_ssl=True)
5. In-Cluster Config: Use service accounts for in-cluster applications

• GitHub: https://github.com/kubernetes-client/python
• Documentation: https://kubernetes.readthedocs.io/
• PyPI: https://pypi.org/project/kubernetes/
• Examples: https://github.com/kubernetes-client/python/tree/master/examples