kubernetes-python（Kubernetes Python 测试）

Kubernetes Python Client Skill

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

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:

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

Security Best Practices

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

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