tessl/pypi-kubernetes-asyncio

Asynchronous Python client library for the Kubernetes API providing async/await support for all Kubernetes operations

—

Pending

Overview

Eval results

Files

Client APIs

Name: tessl/pypi-kubernetes-asyncio
Author: tessl

Comprehensive set of auto-generated API classes providing async access to all Kubernetes API groups and versions. The client module includes 64 API classes and 694 model classes covering the complete Kubernetes API surface.

Capabilities

Core API Client

The foundation ApiClient class that handles HTTP communication, authentication, serialization, and deserialization for all Kubernetes API operations.

class ApiClient:
    def __init__(self, configuration=None, header_name=None, header_value=None, 
                 cookie=None, pool_threads=1):
        """
        Initialize API client.
        
        Parameters:
        - configuration: Configuration object with cluster details
        - header_name: Custom header name for authentication
        - header_value: Custom header value for authentication  
        - cookie: Cookie string for authentication
        - pool_threads: Number of threads for connection pooling
        """
    
    async def __aenter__(self):
        """Async context manager entry."""
        
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit with proper cleanup."""
        
    async def close(self):
        """Close all connections and clean up resources."""
        
    def call_api(self, resource_path, method, path_params=None, query_params=None,
                header_params=None, body=None, post_params=None, files=None,
                response_type=None, auth_settings=None, collection_formats=None,
                _preload_content=True, _request_timeout=None):
        """Make HTTP API call with automatic serialization/deserialization."""

Configuration

Client configuration management for cluster connection details, authentication, and behavior settings.

class Configuration:
    def __init__(self):
        """Initialize default configuration."""
        
    # Connection settings
    host: str  # Kubernetes API server URL
    ssl_ca_cert: str  # Path to CA certificate file
    cert_file: str  # Path to client certificate file
    key_file: str  # Path to client private key file
    verify_ssl: bool  # Whether to verify SSL certificates
    
    # Authentication settings  
    api_key: dict  # API key authentication mapping
    api_key_prefix: dict  # API key prefix mapping
    username: str  # Basic auth username
    password: str  # Basic auth password
    
    # Client behavior
    client_side_validation: bool  # Enable client-side validation
    discard_unknown_keys: bool  # Ignore unknown properties in responses
    timeout: int  # Request timeout in seconds

Core V1 API

Primary API for managing core Kubernetes resources like pods, services, namespaces, nodes, secrets, and config maps.

class CoreV1Api:
    def __init__(self, api_client=None):
        """Initialize Core V1 API client."""
    
    # Pod operations
    async def list_pod_for_all_namespaces(self, **kwargs):
        """List all pods across all namespaces."""
        
    async def list_namespaced_pod(self, namespace: str, **kwargs):
        """List pods in a specific namespace."""
        
    async def create_namespaced_pod(self, namespace: str, body: V1Pod, **kwargs):
        """Create a new pod in the specified namespace."""
        
    async def read_namespaced_pod(self, name: str, namespace: str, **kwargs):
        """Read details of a specific pod."""
        
    async def patch_namespaced_pod(self, name: str, namespace: str, body: object, **kwargs):
        """Partially update a pod."""
        
    async def delete_namespaced_pod(self, name: str, namespace: str, **kwargs):
        """Delete a pod."""
        
    # Service operations
    async def list_namespaced_service(self, namespace: str, **kwargs):
        """List services in a namespace."""
        
    async def create_namespaced_service(self, namespace: str, body: V1Service, **kwargs):
        """Create a new service."""
        
    async def read_namespaced_service(self, name: str, namespace: str, **kwargs):
        """Read service details."""
        
    async def patch_namespaced_service(self, name: str, namespace: str, body: object, **kwargs):
        """Update a service."""
        
    async def delete_namespaced_service(self, name: str, namespace: str, **kwargs):
        """Delete a service."""
        
    # Namespace operations
    async def list_namespace(self, **kwargs):
        """List all namespaces."""
        
    async def create_namespace(self, body: V1Namespace, **kwargs):
        """Create a new namespace."""
        
    async def read_namespace(self, name: str, **kwargs):
        """Read namespace details."""
        
    async def delete_namespace(self, name: str, **kwargs):
        """Delete a namespace."""
        
    # Node operations
    async def list_node(self, **kwargs):
        """List all nodes in the cluster."""
        
    async def read_node(self, name: str, **kwargs):
        """Read node details."""
        
    async def patch_node(self, name: str, body: object, **kwargs):
        """Update node labels, annotations, or spec."""
        
    # Secret operations
    async def list_namespaced_secret(self, namespace: str, **kwargs):
        """List secrets in a namespace."""
        
    async def create_namespaced_secret(self, namespace: str, body: V1Secret, **kwargs):
        """Create a new secret."""
        
    async def read_namespaced_secret(self, name: str, namespace: str, **kwargs):
        """Read secret details."""
        
    async def patch_namespaced_secret(self, name: str, namespace: str, body: object, **kwargs):
        """Update a secret."""
        
    async def delete_namespaced_secret(self, name: str, namespace: str, **kwargs):
        """Delete a secret."""
        
    # ConfigMap operations  
    async def list_namespaced_config_map(self, namespace: str, **kwargs):
        """List config maps in a namespace."""
        
    async def create_namespaced_config_map(self, namespace: str, body: V1ConfigMap, **kwargs):
        """Create a new config map."""
        
    async def read_namespaced_config_map(self, name: str, namespace: str, **kwargs):
        """Read config map details."""
        
    async def patch_namespaced_config_map(self, name: str, namespace: str, body: object, **kwargs):
        """Update a config map."""
        
    async def delete_namespaced_config_map(self, name: str, namespace: str, **kwargs):
        """Delete a config map."""

Apps V1 API

API for managing application workloads including deployments, replica sets, daemon sets, and stateful sets.

class AppsV1Api:
    def __init__(self, api_client=None):
        """Initialize Apps V1 API client."""
    
    # Deployment operations
    async def list_namespaced_deployment(self, namespace: str, **kwargs):
        """List deployments in a namespace."""
        
    async def create_namespaced_deployment(self, namespace: str, body: V1Deployment, **kwargs):
        """Create a new deployment."""
        
    async def read_namespaced_deployment(self, name: str, namespace: str, **kwargs):
        """Read deployment details."""
        
    async def patch_namespaced_deployment(self, name: str, namespace: str, body: object, **kwargs):
        """Update a deployment."""
        
    async def delete_namespaced_deployment(self, name: str, namespace: str, **kwargs):
        """Delete a deployment."""
        
    # ReplicaSet operations
    async def list_namespaced_replica_set(self, namespace: str, **kwargs):
        """List replica sets in a namespace."""
        
    async def create_namespaced_replica_set(self, namespace: str, body: V1ReplicaSet, **kwargs):
        """Create a new replica set."""
        
    # DaemonSet operations
    async def list_namespaced_daemon_set(self, namespace: str, **kwargs):
        """List daemon sets in a namespace."""
        
    async def create_namespaced_daemon_set(self, namespace: str, body: V1DaemonSet, **kwargs):
        """Create a new daemon set."""
        
    # StatefulSet operations
    async def list_namespaced_stateful_set(self, namespace: str, **kwargs):
        """List stateful sets in a namespace."""
        
    async def create_namespaced_stateful_set(self, namespace: str, body: V1StatefulSet, **kwargs):
        """Create a new stateful set."""

Batch V1 API

API for managing batch workloads including jobs and cron jobs.

class BatchV1Api:
    def __init__(self, api_client=None):
        """Initialize Batch V1 API client."""
    
    # Job operations
    async def list_namespaced_job(self, namespace: str, **kwargs):
        """List jobs in a namespace."""
        
    async def create_namespaced_job(self, namespace: str, body: V1Job, **kwargs):
        """Create a new job."""
        
    async def read_namespaced_job(self, name: str, namespace: str, **kwargs):
        """Read job details."""
        
    async def delete_namespaced_job(self, name: str, namespace: str, **kwargs):
        """Delete a job."""
        
    # CronJob operations
    async def list_namespaced_cron_job(self, namespace: str, **kwargs):
        """List cron jobs in a namespace."""
        
    async def create_namespaced_cron_job(self, namespace: str, body: V1CronJob, **kwargs):
        """Create a new cron job."""
        
    async def read_namespaced_cron_job(self, name: str, namespace: str, **kwargs):
        """Read cron job details."""
        
    async def patch_namespaced_cron_job(self, name: str, namespace: str, body: object, **kwargs):
        """Update a cron job."""
        
    async def delete_namespaced_cron_job(self, name: str, namespace: str, **kwargs):
        """Delete a cron job."""

Custom Objects API

API for managing custom resources defined by Custom Resource Definitions (CRDs).

class CustomObjectsApi:
    def __init__(self, api_client=None):
        """Initialize Custom Objects API client."""
    
    async def list_cluster_custom_object(self, group: str, version: str, plural: str, **kwargs):
        """List custom objects across all namespaces."""
        
    async def list_namespaced_custom_object(self, group: str, version: str, namespace: str, 
                                           plural: str, **kwargs):
        """List custom objects in a specific namespace."""
        
    async def create_namespaced_custom_object(self, group: str, version: str, namespace: str,
                                             plural: str, body: object, **kwargs):
        """Create a custom object."""
        
    async def get_namespaced_custom_object(self, group: str, version: str, namespace: str,
                                          plural: str, name: str, **kwargs):
        """Get a specific custom object."""
        
    async def patch_namespaced_custom_object(self, group: str, version: str, namespace: str,
                                            plural: str, name: str, body: object, **kwargs):
        """Update a custom object."""
        
    async def delete_namespaced_custom_object(self, group: str, version: str, namespace: str,
                                             plural: str, name: str, **kwargs):
        """Delete a custom object."""

Key Model Classes

Core Resource Models

class V1Pod:
    def __init__(self, api_version=None, kind=None, metadata=None, spec=None, status=None):
        """Pod resource model."""
        
class V1PodSpec:
    def __init__(self, containers, volumes=None, restart_policy=None, node_name=None, 
                 service_account_name=None, **kwargs):
        """Pod specification."""
        
class V1Container:
    def __init__(self, name: str, image: str, ports=None, env=None, volume_mounts=None,
                 resources=None, **kwargs):
        """Container specification."""
        
class V1Service:
    def __init__(self, api_version=None, kind=None, metadata=None, spec=None, status=None):
        """Service resource model."""
        
class V1ServiceSpec:
    def __init__(self, ports=None, selector=None, type=None, cluster_ip=None, **kwargs):
        """Service specification."""
        
class V1Deployment:
    def __init__(self, api_version=None, kind=None, metadata=None, spec=None, status=None):
        """Deployment resource model."""
        
class V1DeploymentSpec:
    def __init__(self, replicas=None, selector=None, template=None, strategy=None, **kwargs):
        """Deployment specification."""

Common Types

class V1ObjectMeta:
    def __init__(self, name=None, namespace=None, labels=None, annotations=None, **kwargs):
        """Object metadata common to all resources."""
        
class V1LabelSelector:
    def __init__(self, match_labels=None, match_expressions=None):
        """Label selector for identifying resources."""
        
class V1Status:
    def __init__(self, kind=None, api_version=None, metadata=None, status=None,
                 message=None, reason=None, code=None, **kwargs):
        """Status response from API operations."""
        
class V1DeleteOptions:
    def __init__(self, grace_period_seconds=None, propagation_policy=None, **kwargs):
        """Options for delete operations."""

Usage Examples

Basic Resource Management

import asyncio
from kubernetes_asyncio import client, config

async def manage_pods():
    await config.load_config()
    v1 = client.CoreV1Api()
    
    try:
        # Create a pod
        pod = client.V1Pod(
            api_version="v1",
            kind="Pod", 
            metadata=client.V1ObjectMeta(name="test-pod", namespace="default"),
            spec=client.V1PodSpec(
                containers=[
                    client.V1Container(
                        name="test-container",
                        image="nginx:latest",
                        ports=[client.V1ContainerPort(container_port=80)]
                    )
                ]
            )
        )
        
        result = await v1.create_namespaced_pod(namespace="default", body=pod)
        print(f"Pod created: {result.metadata.name}")
        
        # List pods
        pod_list = await v1.list_namespaced_pod(namespace="default")
        for pod in pod_list.items:
            print(f"Found pod: {pod.metadata.name} - {pod.status.phase}")
            
    finally:
        await v1.api_client.close()

asyncio.run(manage_pods())

Working with Deployments

async def manage_deployment():
    await config.load_config()
    apps_v1 = client.AppsV1Api()
    
    try:
        # Create deployment
        deployment = client.V1Deployment(
            api_version="apps/v1",
            kind="Deployment",
            metadata=client.V1ObjectMeta(name="nginx-deployment", namespace="default"),
            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.20",
                                ports=[client.V1ContainerPort(container_port=80)]
                            )
                        ]
                    )
                )
            )
        )
        
        result = await apps_v1.create_namespaced_deployment(namespace="default", body=deployment)
        print(f"Deployment created: {result.metadata.name}")
        
    finally:
        await apps_v1.api_client.close()

Install with Tessl CLI

npx tessl i tessl/pypi-kubernetes-asyncio

docs