or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

application-workloads.mdautoscaling.mdconfiguration.mdcore-resources.mdcustom-resources.mddynamic-client.mdindex.mdleader-election.mdnetworking.mdrbac-security.mdresource-watching.mdstorage.mdstreaming-operations.mdutilities.md
tile.json

tessl/pypi-kubernetes

Python client library for interacting with Kubernetes clusters through the Kubernetes API

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/kubernetes@33.1.x

To install, run

npx @tessl/cli install tessl/pypi-kubernetes@33.1.0

index.mddocs/

Kubernetes Python Client

The official Python client library for Kubernetes, providing comprehensive programmatic access to Kubernetes clusters through the Kubernetes API. This library enables developers to manage all Kubernetes resources including pods, services, deployments, and custom resources with full support for authentication, configuration management, and dynamic client capabilities.

Package Information

  • Package Name: kubernetes
  • Language: Python
  • Installation: pip install kubernetes

Core Imports

from kubernetes import client, config

For dynamic client operations:

from kubernetes import dynamic

For watching resource changes:

from kubernetes import watch

For streaming operations (exec, logs, port-forward):

from kubernetes import stream

For utility functions:

from kubernetes import utils

For leader election:

from kubernetes import leaderelection

For networking operations:

from kubernetes.client import NetworkingV1Api

For storage management:

from kubernetes.client import StorageV1Api

For autoscaling:

from kubernetes.client import AutoscalingV1Api, AutoscalingV2Api

Basic Usage

from kubernetes import client, config

# Load kubeconfig from default location or in-cluster config
config.load_kube_config()  # or config.load_incluster_config()

# Create API client instance
v1 = client.CoreV1Api()

# List all pods in default namespace
pods = v1.list_namespaced_pod(namespace="default")
for pod in pods.items:
    print(f"Pod: {pod.metadata.name}")

# Create a simple pod
pod_manifest = {
    "apiVersion": "v1",
    "kind": "Pod",
    "metadata": {"name": "test-pod"},
    "spec": {
        "containers": [{
            "name": "test-container",
            "image": "nginx:latest",
            "ports": [{"containerPort": 80}]
        }]
    }
}

# Create the pod
v1.create_namespaced_pod(namespace="default", body=pod_manifest)

Architecture

The Kubernetes Python client is organized around several key components:

  • Generated API Classes: Auto-generated client classes for all Kubernetes API groups (CoreV1Api, AppsV1Api, etc.)
  • Model Classes: Python representations of all Kubernetes resources (V1Pod, V1Service, etc.)
  • Configuration: Support for kubeconfig files and in-cluster authentication
  • Dynamic Client: Generic client for any Kubernetes resource without pre-generated classes
  • Watch: Event streaming for real-time resource monitoring
  • Stream: Execute commands and stream logs from pods
  • Utils: Helper functions for common operations like creating resources from YAML

Capabilities

Configuration and Authentication

Handles loading Kubernetes cluster configuration from kubeconfig files, in-cluster service accounts, or programmatic configuration, with support for various authentication methods including certificates, tokens, and cloud provider authentication.

def load_kube_config(config_file=None, context=None, client_configuration=None, persist_config=True): ...
def load_incluster_config(): ...
def list_kube_config_contexts(config_file=None): ...
def new_client_from_config(config_file=None, context=None, persist_config=True): ...

Configuration

Core API Resources

Management of fundamental Kubernetes resources including pods, services, namespaces, config maps, secrets, and persistent volumes. These form the basic building blocks for all Kubernetes applications.

class CoreV1Api:
    def create_namespaced_pod(self, namespace: str, body: V1Pod, **kwargs) -> V1Pod: ...
    def list_namespaced_pod(self, namespace: str, **kwargs) -> V1PodList: ...
    def read_namespaced_pod(self, name: str, namespace: str, **kwargs) -> V1Pod: ...
    def delete_namespaced_pod(self, name: str, namespace: str, **kwargs) -> V1Status: ...
    def create_namespaced_service(self, namespace: str, body: V1Service, **kwargs) -> V1Service: ...
    def list_namespaced_service(self, namespace: str, **kwargs) -> V1ServiceList: ...

Core Resources

Application Workloads

Management of application deployment patterns including deployments, replica sets, daemon sets, stateful sets, jobs, and cron jobs. These resources provide scalable and reliable application hosting patterns.

class AppsV1Api:
    def create_namespaced_deployment(self, namespace: str, body: V1Deployment, **kwargs) -> V1Deployment: ...
    def list_namespaced_deployment(self, namespace: str, **kwargs) -> V1DeploymentList: ...
    def patch_namespaced_deployment(self, name: str, namespace: str, body: object, **kwargs) -> V1Deployment: ...
    def create_namespaced_stateful_set(self, namespace: str, body: V1StatefulSet, **kwargs) -> V1StatefulSet: ...
    def create_namespaced_daemon_set(self, namespace: str, body: V1DaemonSet, **kwargs) -> V1DaemonSet: ...

Application Workloads

Dynamic Client Operations

Generic client for working with any Kubernetes resource without pre-generated classes. Enables operations on custom resources, API discovery, and flexible resource manipulation with runtime type resolution.

class DynamicClient:
    def __init__(self, client: ApiClient, cache_file: str = None, discoverer=None): ...
    def get(self, resource: Resource, name: str = None, namespace: str = None, **kwargs) -> ResourceInstance: ...
    def create(self, resource: Resource, body: dict = None, **kwargs) -> ResourceInstance: ...
    def delete(self, resource: Resource, name: str = None, namespace: str = None, **kwargs) -> ResourceInstance: ...
    def patch(self, resource: Resource, body: dict = None, **kwargs) -> ResourceInstance: ...

Dynamic Client

Resource Watching

Event streaming capabilities for monitoring real-time changes to Kubernetes resources. Provides efficient watch operations that can be filtered and processed for automated responses to cluster state changes.

class Watch:
    def stream(self, func, *args, **kwargs): ...
    def stop(self): ...

Resource Watching

Pod Streaming Operations

Execute commands in running pods, stream container logs, and establish port forwarding connections. Essential for debugging, log analysis, and establishing secure connections to pod services.

def stream(ws_client, channel, *args, **kwargs): ...
def portforward(api_instance, name: str, namespace: str, ports: list, **kwargs): ...

Streaming Operations

Utility Functions

Helper functions for common Kubernetes operations including creating resources from YAML files, parsing quantity values, and working with Kubernetes-specific data formats.

def create_from_yaml(k8s_client, yaml_file: str, namespace: str = "default", **kwargs): ...
def create_from_dict(k8s_client, data: dict, namespace: str = "default", **kwargs): ...
def parse_quantity(quantity: str) -> float: ...
def parse_duration(duration: str) -> int: ...

Utilities

Custom Resources and API Extensions

Work with custom resource definitions, API extensions, and admission controllers. Enables extending Kubernetes with custom APIs and controllers while maintaining full client library support.

class ApiextensionsV1Api:
    def create_custom_resource_definition(self, body: V1CustomResourceDefinition, **kwargs) -> V1CustomResourceDefinition: ...
    def list_custom_resource_definition(self, **kwargs) -> V1CustomResourceDefinitionList: ...

class CustomObjectsApi:
    def create_namespaced_custom_object(self, group: str, version: str, namespace: str, plural: str, body: object, **kwargs) -> object: ...
    def list_namespaced_custom_object(self, group: str, version: str, namespace: str, plural: str, **kwargs) -> object: ...

Custom Resources

RBAC and Security

Role-based access control management including roles, cluster roles, role bindings, service accounts, and security policies. Essential for implementing proper access controls and security policies in Kubernetes clusters.

class RbacAuthorizationV1Api:
    def create_namespaced_role(self, namespace: str, body: V1Role, **kwargs) -> V1Role: ...
    def create_cluster_role(self, body: V1ClusterRole, **kwargs) -> V1ClusterRole: ...
    def create_namespaced_role_binding(self, namespace: str, body: V1RoleBinding, **kwargs) -> V1RoleBinding: ...

RBAC and Security

Leader Election

Distributed leader election for building highly available applications where only one instance should be active. Uses Kubernetes resources as coordination locks with automatic lease renewal and failover capabilities.

class LeaderElection:
    def __init__(self, election_config: Config): ...
    def run(self) -> None: ...

class Config:
    def __init__(self, lock, lease_duration: int, renew_deadline: int, retry_period: int, onstarted_leading: callable, onstopped_leading: callable = None): ...

class ConfigMapLock:
    def __init__(self, name: str, namespace: str, identity: str): ...

Leader Election

Networking

Network management including ingress controllers, network policies, and service networking. Provides control over HTTP/HTTPS routing, traffic policies, and network security within clusters.

class NetworkingV1Api:
    def create_namespaced_ingress(self, namespace: str, body: V1Ingress, **kwargs) -> V1Ingress: ...
    def create_namespaced_network_policy(self, namespace: str, body: V1NetworkPolicy, **kwargs) -> V1NetworkPolicy: ...
    def create_ingress_class(self, body: V1IngressClass, **kwargs) -> V1IngressClass: ...

Networking

Storage Management

Advanced storage capabilities including storage classes, volume attachments, CSI drivers, and storage capacity management for persistent storage infrastructure.

class StorageV1Api:
    def create_storage_class(self, body: V1StorageClass, **kwargs) -> V1StorageClass: ...
    def create_volume_attachment(self, body: V1VolumeAttachment, **kwargs) -> V1VolumeAttachment: ...
    def create_csi_driver(self, body: V1CSIDriver, **kwargs) -> V1CSIDriver: ...

Storage Management

Autoscaling

Horizontal Pod Autoscaler (HPA) management for automatic scaling based on CPU, memory, or custom metrics with advanced scaling behaviors and policies.

class AutoscalingV1Api:
    def create_namespaced_horizontal_pod_autoscaler(self, namespace: str, body: V1HorizontalPodAutoscaler, **kwargs) -> V1HorizontalPodAutoscaler: ...

class AutoscalingV2Api:
    def create_namespaced_horizontal_pod_autoscaler(self, namespace: str, body: V2HorizontalPodAutoscaler, **kwargs) -> V2HorizontalPodAutoscaler: ...

Autoscaling

Common Resource Types

V1Pod

class V1Pod:
    api_version: str
    kind: str
    metadata: V1ObjectMeta
    spec: V1PodSpec
    status: V1PodStatus

V1Service

class V1Service:
    api_version: str
    kind: str
    metadata: V1ObjectMeta
    spec: V1ServiceSpec
    status: V1ServiceStatus

V1Deployment

class V1Deployment:
    api_version: str
    kind: str
    metadata: V1ObjectMeta
    spec: V1DeploymentSpec
    status: V1DeploymentStatus

V1ObjectMeta

class V1ObjectMeta:
    name: str
    namespace: str
    labels: dict
    annotations: dict
    creation_timestamp: datetime
    uid: str
    resource_version: str

Exception Types

class ApiException(Exception):
    status: int
    reason: str
    body: str
    headers: dict

class ConfigException(Exception):
    pass

class FailToCreateError(Exception):
    api_exceptions: list