kubernetes-python

Kubernetes Python client for programmatic cluster management. Use when working with Kubernetes API, managing pods, deployments, services, namespaces, configmaps, secrets, jobs, CRDs, EKS clusters, watching resources, automating K8s operations, or building Kubernetes controllers.

Quality

90%

Does it follow best practices?

Run evals on this skill

Adds up to 20 points to the overall score

View guide

Securityby

Passed

No findings from the security scan

Kubernetes Python Client Skill

Name: kubernetes-python
Rating: 73.6 (1 reviews)
Author: fernandezbaptiste

Official Python client library for Kubernetes, providing programmatic access to the Kubernetes API for automation, custom tooling, and application integration.

Quick Start

Installation

pip install kubernetes

Basic Usage

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}")

Core Concepts

Client Initialization Patterns

Local Development (using kubeconfig):

from kubernetes import client, config

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

In-Cluster (running inside Kubernetes):

from kubernetes import client, config

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

Specific Context:

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()`

Common Operations

Creating a Deployment

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

# 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):

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):

deployment.metadata.resource_version = existing.metadata.resource_version
apps_v1.replace_namespaced_deployment(
    name="nginx-deployment",
    namespace="default",
    body=deployment
)

Deleting Resources

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
)

Error Handling

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)

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 Resources

Watch for resource changes in real-time:

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

Working with ConfigMaps and Secrets

ConfigMap

configmap = client.V1ConfigMap(
    metadata=client.V1ObjectMeta(name="my-config"),
    data={"key1": "value1", "key2": "value2"}
)

v1.create_namespaced_config_map(
    namespace="default",
    body=configmap
)

Secret

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
)

Custom Resources (CRDs)

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
)

Production Patterns

Timeout Configuration

# Set timeout for operations
pods = v1.list_namespaced_pod(
    namespace="default",
    _request_timeout=10  # 10 second timeout
)

Pagination for Large Lists

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

# 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']

Reference Documentation

For detailed information, see:

Client Setup: Authentication methods, EKS integration, configuration
Resource Operations: Comprehensive CRUD patterns for all resource types
Watch Patterns: Advanced watch patterns, resilient loops, async operations

Key Features

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)

Version Compatibility

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

Security Best Practices

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

Quick Reference Links

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

Repository: fernandezbaptiste/Skrillz
Path: skills/kubernetes-python/SKILL.md
Commit: 93ed392

Last updated: 2 days ago
First committed: 8 months ago

Also appears in

fernandezbaptiste/Skrillz

In sync

since Sep 12, 2026

Is this your skill?

If you maintain this skill, you can claim it as your own. Once claimed, you can manage eval scenarios, bundle related skills, attach documentation or rules, and ensure cross-agent compatibility.