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

Configuration

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

Multiple methods for loading Kubernetes cluster credentials and configuration, supporting kubeconfig files, in-cluster authentication, exec providers, OIDC, and various credential plugins. The configuration system automatically detects the appropriate authentication method and handles credential refresh.

Capabilities

Primary Configuration Functions

The main entry points for loading Kubernetes configuration in different environments.

async def load_config(**kwargs):
    """
    Unified configuration loader that automatically detects and loads appropriate config.
    
    Tries in order:
    1. Provided config_file or kube_config_path parameter
    2. Default kubeconfig location (~/.kube/config)  
    3. In-cluster configuration (service account)
    
    Parameters:
    - config_file: str, path to kubeconfig file
    - kube_config_path: str, alternative parameter name for config_file
    - context: str, kubeconfig context to use
    - client_configuration: Configuration, object to populate
    - persist_config: bool, whether to set as default global config
    """

async def load_kube_config(config_file=None, context=None, client_configuration=None,
                          persist_config=True, **kwargs):
    """
    Load configuration from kubeconfig file.
    
    Parameters:
    - config_file: str, path to kubeconfig file (default: ~/.kube/config)
    - context: str, kubeconfig context to use (default: current-context)
    - client_configuration: Configuration, object to populate
    - persist_config: bool, whether to set as default global config
    """

def load_incluster_config(client_configuration=None, try_refresh_token=True):
    """
    Load in-cluster configuration from service account mounted in pod.
    
    Uses:
    - /var/run/secrets/kubernetes.io/serviceaccount/token for authentication
    - /var/run/secrets/kubernetes.io/serviceaccount/ca.crt for CA certificate
    - KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT environment variables
    
    Parameters:
    - client_configuration: Configuration, object to populate
    - try_refresh_token: bool, whether to refresh tokens automatically
    """

async def load_kube_config_from_dict(config_dict, context=None, client_configuration=None,
                                     temp_file_path=None):
    """
    Load configuration from dictionary representation of kubeconfig.
    
    Parameters:
    - config_dict: dict, kubeconfig as dictionary
    - context: str, context name to use
    - client_configuration: Configuration, object to populate
    - temp_file_path: str, directory for temporary files
    """

Client Creation Functions

Functions for creating ApiClient instances with pre-configured authentication.

def new_client_from_config(config_file=None, context=None, persist_config=True):
    """
    Create new ApiClient from kubeconfig file.
    
    Parameters:
    - config_file: str, path to kubeconfig file
    - context: str, kubeconfig context to use
    - persist_config: bool, whether to set as default global config
    
    Returns:
    - ApiClient: Configured client ready for use
    """

def new_client_from_config_dict(config_dict, context=None, persist_config=True):
    """
    Create new ApiClient from kubeconfig dictionary.
    
    Parameters:
    - config_dict: dict, kubeconfig as dictionary
    - context: str, context name to use
    - persist_config: bool, whether to set as default global config
    
    Returns:
    - ApiClient: Configured client ready for use
    """

Context Management

Functions for discovering and managing kubeconfig contexts.

def list_kube_config_contexts(config_file=None):
    """
    List all available contexts in kubeconfig file.
    
    Parameters:
    - config_file: str, path to kubeconfig file (default: ~/.kube/config)
    
    Returns:
    - tuple: (contexts_list, current_context_dict)
      - contexts_list: List of all context dictionaries
      - current_context_dict: Currently active context dictionary
    """

Configuration Classes

Core classes for managing configuration state and loading kubeconfig files.

class KubeConfigLoader:
    def __init__(self, config_dict, active_context=None, get_google_credentials=None,
                 config_base_path=None, config_persister=None):
        """
        Kubeconfig file parser and processor.
        
        Parameters:
        - config_dict: dict, kubeconfig content as dictionary
        - active_context: str, context name to use
        - get_google_credentials: callable, Google Cloud credential provider
        - config_base_path: str, base path for resolving relative file paths
        - config_persister: callable, function to persist configuration changes
        """
    
    def load_and_set(self, client_configuration):
        """Load kubeconfig and populate Configuration object."""
        
    def _load_authentication(self):
        """Load authentication configuration for active context."""
        
    def _load_cluster_info(self):
        """Load cluster connection information."""

class FileOrData:
    def __init__(self, data=None, file_path=None, file_base_path=None):
        """
        Utility for handling kubeconfig fields that can be inline data or file references.
        
        Parameters:
        - data: str, inline data content
        - file_path: str, path to file containing data
        - file_base_path: str, base path for resolving relative file paths
        """
    
    def as_data(self):
        """Get data content as string."""
        
    def as_bytes(self):
        """Get data content as bytes."""

class InClusterConfigLoader:
    def __init__(self, token_filename=None, cert_filename=None, environ=None):
        """
        In-cluster configuration loader for pods.
        
        Parameters:
        - token_filename: str, path to service account token file
        - cert_filename: str, path to CA certificate file  
        - environ: dict, environment variables
        """
    
    def load_and_set(self, client_configuration):
        """Load in-cluster config and populate Configuration object."""

Authentication Providers

Support for various Kubernetes authentication mechanisms.

class ExecProvider:
    def __init__(self, exec_config):
        """
        External credential provider (exec) support.
        
        Parameters:
        - exec_config: dict, exec configuration from kubeconfig
        """
    
    def run(self, previous_response=None):
        """Execute credential provider and return credentials."""

class OpenIDRequestor:
    def __init__(self, configuration, id_token_string):
        """
        OpenID Connect (OIDC) authentication support.
        
        Parameters:
        - configuration: Configuration, client configuration
        - id_token_string: str, OIDC ID token
        """
    
    def refresh_oidc_token(self):
        """Refresh OIDC token using refresh token."""

Exception Classes

Configuration-specific exception types for error handling.

class ConfigException(Exception):
    """
    Configuration-related errors including file parsing, authentication failures,
    and invalid kubeconfig structure.
    """
    def __init__(self, message):
        """Initialize with error message."""

Constants

Important constants used throughout the configuration system.

# Default kubeconfig file location
KUBE_CONFIG_DEFAULT_LOCATION: str = "~/.kube/config"

# In-cluster service account environment variables
SERVICE_HOST_ENV_NAME: str = "KUBERNETES_SERVICE_HOST"
SERVICE_PORT_ENV_NAME: str = "KUBERNETES_SERVICE_PORT"

# In-cluster service account file paths
SERVICE_TOKEN_FILENAME: str = "/var/run/secrets/kubernetes.io/serviceaccount/token"
SERVICE_CERT_FILENAME: str = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
SERVICE_NAMESPACE_FILENAME: str = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"

Usage Examples

Basic Configuration Loading

import asyncio
from kubernetes_asyncio import client, config

async def basic_config():
    # Automatically detect and load appropriate configuration
    await config.load_config()
    
    # Use the loaded configuration
    v1 = client.CoreV1Api()
    pods = await v1.list_pod_for_all_namespaces()
    print(f"Found {len(pods.items)} pods")
    await v1.api_client.close()

asyncio.run(basic_config())

Specific Configuration File

async def specific_config():
    # Load specific kubeconfig file and context
    await config.load_kube_config(
        config_file="/path/to/custom/kubeconfig",
        context="production-cluster"
    )
    
    v1 = client.CoreV1Api()
    namespaces = await v1.list_namespace()
    print(f"Cluster has {len(namespaces.items)} namespaces")
    await v1.api_client.close()

In-Cluster Configuration

async def incluster_config():
    # For applications running inside Kubernetes pods
    try:
        config.load_incluster_config()
        print("Using in-cluster configuration")
    except config.ConfigException:
        # Fallback to kubeconfig file
        await config.load_kube_config()
        print("Using kubeconfig file")
    
    v1 = client.CoreV1Api()
    # ... use client
    await v1.api_client.close()

Multiple Contexts

async def multiple_contexts():
    # List available contexts
    contexts, active_context = config.list_kube_config_contexts()
    
    print("Available contexts:")
    for context in contexts:
        name = context['name']
        cluster = context['context']['cluster']
        user = context['context']['user']
        print(f"  {name}: cluster={cluster}, user={user}")
    
    print(f"Current context: {active_context['name']}")
    
    # Switch to different context
    await config.load_kube_config(context="staging-cluster")

Custom Configuration Object

async def custom_config():
    # Create custom configuration
    configuration = client.Configuration()
    
    # Load into custom configuration object
    await config.load_kube_config(client_configuration=configuration)
    
    # Use with specific client
    api_client = client.ApiClient(configuration)
    v1 = client.CoreV1Api(api_client)
    
    try:
        nodes = await v1.list_node()
        print(f"Cluster has {len(nodes.items)} nodes")
    finally:
        await api_client.close()

Client Factory Pattern

async def client_factory():
    # Create pre-configured clients
    api_client = config.new_client_from_config(
        config_file="/path/to/kubeconfig",
        context="production"
    )
    
    # Use with any API
    v1 = client.CoreV1Api(api_client)
    apps_v1 = client.AppsV1Api(api_client)
    
    try:
        pods = await v1.list_pod_for_all_namespaces()
        deployments = await apps_v1.list_deployment_for_all_namespaces()
        print(f"Found {len(pods.items)} pods and {len(deployments.items)} deployments")
    finally:
        await api_client.close()

Programmatic Configuration

async def programmatic_config():
    # Build configuration from environment or secrets
    configuration = client.Configuration()
    configuration.host = "https://kubernetes.example.com:6443"
    configuration.api_key = {"authorization": "Bearer " + token}
    configuration.ssl_ca_cert = "/path/to/ca.crt"
    
    # Create client with custom configuration
    api_client = client.ApiClient(configuration)
    v1 = client.CoreV1Api(api_client)
    
    try:
        # Test connection
        version = await client.VersionApi(api_client).get_code()
        print(f"Connected to Kubernetes {version.git_version}")
    finally:
        await api_client.close()

Install with Tessl CLI

npx tessl i tessl/pypi-kubernetes-asyncio

docs