tessl/pypi-kubernetes

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

—

Pending

Overview

Eval results

Files

Dynamic Client Operations

Name: tessl/pypi-kubernetes
Author: tessl

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. Ideal for working with resources that may not have static client support.

Capabilities

Dynamic Client Creation

Create a dynamic client that can work with any Kubernetes resource by discovering API structure at runtime.

class DynamicClient:
    def __init__(
        self,
        client: ApiClient,
        cache_file: str = None,
        discoverer = None
    ):
        """
        Create dynamic client instance.
        
        Parameters:
        - client: Configured ApiClient instance
        - cache_file: File to cache API discovery information
        - discoverer: Custom discoverer instance (EagerDiscoverer or LazyDiscoverer)
        """

Resource Operations

Perform CRUD operations on any Kubernetes resource using generic methods.

class DynamicClient:
    def get(
        self,
        resource: Resource,
        name: str = None,
        namespace: str = None,
        label_selector: str = None,
        field_selector: str = None,
        **kwargs
    ) -> ResourceInstance:
        """
        Get resource(s) by name or list all resources.
        
        Parameters:
        - resource: Resource definition
        - name: Resource name (if None, lists all resources)
        - namespace: Namespace (for namespaced resources)
        - label_selector: Label selector filter
        - field_selector: Field selector filter
        
        Returns:
        ResourceInstance or ResourceList
        """
    
    def create(
        self,
        resource: Resource,
        body: dict = None,
        namespace: str = None,
        **kwargs
    ) -> ResourceInstance:
        """
        Create a new resource.
        
        Parameters:
        - resource: Resource definition
        - body: Resource specification as dictionary
        - namespace: Namespace (for namespaced resources)
        
        Returns:
        Created ResourceInstance
        """
    
    def delete(
        self,
        resource: Resource,
        name: str = None,
        namespace: str = None,
        body: dict = None,
        label_selector: str = None,
        field_selector: str = None,
        **kwargs
    ) -> ResourceInstance:
        """
        Delete resource(s).
        
        Parameters:
        - resource: Resource definition
        - name: Resource name (if None, uses selectors)
        - namespace: Namespace (for namespaced resources)
        - body: Delete options
        - label_selector: Label selector for bulk delete
        - field_selector: Field selector for bulk delete
        
        Returns:
        Status or deleted ResourceInstance
        """
    
    def patch(
        self,
        resource: Resource,
        body: dict = None,
        name: str = None,
        namespace: str = None,
        **kwargs
    ) -> ResourceInstance:
        """
        Patch (partially update) a resource.
        
        Parameters:
        - resource: Resource definition
        - body: Patch data as dictionary
        - name: Resource name
        - namespace: Namespace (for namespaced resources)
        
        Returns:
        Patched ResourceInstance
        """
    
    def replace(
        self,
        resource: Resource,
        body: dict = None,
        name: str = None,
        namespace: str = None,
        **kwargs
    ) -> ResourceInstance:
        """
        Replace (fully update) a resource.
        
        Parameters:
        - resource: Resource definition
        - body: Complete resource specification
        - name: Resource name
        - namespace: Namespace (for namespaced resources)
        
        Returns:
        Replaced ResourceInstance
        """

Resource Discovery

Discover and access API resources dynamically.

class DynamicClient:
    @property
    def resources(self) -> ResourceRegistry:
        """Access to resource registry for API discovery."""
    
    def ensure_namespace(
        self,
        resource: Resource,
        namespace: str,
        body: dict
    ) -> dict:
        """Ensure namespace is set in resource body if needed."""

API Discovery Classes

Control how API resources are discovered and cached.

class EagerDiscoverer:
    """Preloads all API resources at initialization for faster access."""
    def __init__(self, client: ApiClient): ...

class LazyDiscoverer:
    """Discovers API resources on-demand as they are accessed."""
    def __init__(self, client: ApiClient): ...

Resource Models

Resource

class Resource:
    """Represents a Kubernetes API resource."""
    group: str
    version: str
    kind: str
    name: str  # Plural name
    singular_name: str
    namespaced: bool
    api_version: str
    client: DynamicClient
    
    def create(self, body: dict = None, namespace: str = None, **kwargs) -> ResourceInstance: ...
    def get(self, name: str = None, namespace: str = None, **kwargs) -> ResourceInstance: ...
    def delete(self, name: str = None, namespace: str = None, **kwargs) -> ResourceInstance: ...
    def patch(self, body: dict = None, name: str = None, namespace: str = None, **kwargs) -> ResourceInstance: ...
    def replace(self, body: dict = None, name: str = None, namespace: str = None, **kwargs) -> ResourceInstance: ...

ResourceInstance

class ResourceInstance:
    """Instance of a Kubernetes resource with dynamic attributes."""
    
    # Dynamic attributes based on resource structure
    api_version: str
    kind: str
    metadata: dict
    spec: dict
    status: dict
    
    def to_dict(self) -> dict: ...
    def to_str(self) -> str: ...

ResourceList

class ResourceList:
    """List of ResourceInstance objects."""
    api_version: str
    kind: str
    metadata: dict
    items: list  # List of ResourceInstance
    
    def to_dict(self) -> dict: ...

ResourceField

class ResourceField:
    """Field descriptor for resource attributes."""
    name: str
    type: str
    description: str

Subresource

class Subresource:
    """Represents a subresource (logs, exec, status, etc.)."""
    def get(self, name: str = None, namespace: str = None, **kwargs): ...
    def create(self, body: dict = None, name: str = None, namespace: str = None, **kwargs): ...
    def patch(self, body: dict = None, name: str = None, namespace: str = None, **kwargs): ...
    def replace(self, body: dict = None, name: str = None, namespace: str = None, **kwargs): ...

Usage Examples

Basic Dynamic Client Usage

from kubernetes import client, config, dynamic

# Load configuration
config.load_kube_config()

# Create dynamic client
dyn_client = dynamic.DynamicClient(client.ApiClient())

# Get pods resource
pods = dyn_client.resources.get(api_version="v1", kind="Pod")

# List all pods
pod_list = pods.get()
for pod in pod_list.items:
    print(f"Pod: {pod.metadata.name}")

# Get specific pod
pod = pods.get(name="my-pod", namespace="default")
print(f"Pod status: {pod.status.phase}")

Working with Custom Resources

from kubernetes import client, config, dynamic

config.load_kube_config()
dyn_client = dynamic.DynamicClient(client.ApiClient())

# Access custom resource
my_crds = dyn_client.resources.get(
    api_version="mycompany.io/v1",
    kind="MyCustomResource"
)

# Create custom resource
custom_resource = {
    "apiVersion": "mycompany.io/v1",
    "kind": "MyCustomResource",
    "metadata": {
        "name": "my-resource",
        "namespace": "default"
    },
    "spec": {
        "field1": "value1",
        "field2": "value2"
    }
}

# Create the resource
created = my_crds.create(body=custom_resource, namespace="default")
print(f"Created: {created.metadata.name}")

# List custom resources
resources = my_crds.get()
for resource in resources.items:
    print(f"Custom resource: {resource.metadata.name}")

Resource Discovery

from kubernetes import client, config, dynamic

config.load_kube_config()
dyn_client = dynamic.DynamicClient(client.ApiClient())

# Discover all available resources
for resource in dyn_client.resources.search():
    print(f"Resource: {resource.kind} ({resource.api_version})")
    print(f"  Namespaced: {resource.namespaced}")
    print(f"  Name: {resource.name}")

# Find resources by group
apps_resources = dyn_client.resources.search(group="apps")
for resource in apps_resources:
    print(f"Apps resource: {resource.kind}")

# Find specific resource
deployments = dyn_client.resources.get(api_version="apps/v1", kind="Deployment")
print(f"Found deployments resource: {deployments.name}")

Patching Resources

from kubernetes import client, config, dynamic

config.load_kube_config()
dyn_client = dynamic.DynamicClient(client.ApiClient())

# Get deployments resource
deployments = dyn_client.resources.get(api_version="apps/v1", kind="Deployment")

# Scale deployment using patch
patch_body = {
    "spec": {
        "replicas": 5
    }
}

patched = deployments.patch(
    name="my-deployment",
    namespace="default",
    body=patch_body
)
print(f"Deployment scaled to {patched.spec.replicas} replicas")

# Strategic merge patch for labels
label_patch = {
    "metadata": {
        "labels": {
            "environment": "production",
            "version": "v2.0"
        }
    }
}

deployments.patch(
    name="my-deployment",
    namespace="default",
    body=label_patch
)

Using Different Discoverers

from kubernetes import client, config, dynamic

config.load_kube_config()

# Use eager discoverer (preloads all resources)
eager_discoverer = dynamic.EagerDiscoverer(client.ApiClient())
dyn_client_eager = dynamic.DynamicClient(
    client.ApiClient(),
    discoverer=eager_discoverer
)

# Use lazy discoverer (discovers on-demand)
lazy_discoverer = dynamic.LazyDiscoverer(client.ApiClient())
dyn_client_lazy = dynamic.DynamicClient(
    client.ApiClient(),
    discoverer=lazy_discoverer
)

# Both clients work the same way
pods = dyn_client_eager.resources.get(api_version="v1", kind="Pod")
pod_list = pods.get()

Install with Tessl CLI