tessl/pypi-kubernetes

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

—

Pending

Overview

Eval results

Files

Storage Management

Name: tessl/pypi-kubernetes
Author: tessl

Advanced storage capabilities including storage classes, volume attachments, CSI drivers, and storage capacity management. Provides comprehensive control over persistent storage infrastructure and dynamic volume provisioning.

Capabilities

Storage Classes

Manage storage class definitions for dynamic volume provisioning with different storage backends and policies.

class StorageV1Api:
    def create_storage_class(
        self,
        body: V1StorageClass,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1StorageClass:
        """Create a storage class for dynamic provisioning."""
    
    def list_storage_class(
        self,
        pretty: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1StorageClassList:
        """List storage classes."""
    
    def read_storage_class(
        self,
        name: str,
        pretty: str = None
    ) -> V1StorageClass:
        """Read specified storage class."""
    
    def patch_storage_class(
        self,
        name: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1StorageClass:
        """Update specified storage class."""
    
    def delete_storage_class(
        self,
        name: str,
        dry_run: str = None,
        grace_period_seconds: int = None,
        pretty: str = None
    ) -> V1Status:
        """Delete specified storage class."""

Volume Attachments

Manage volume attachment operations between persistent volumes and nodes.

class StorageV1Api:
    def create_volume_attachment(
        self,
        body: V1VolumeAttachment,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1VolumeAttachment:
        """Create a volume attachment."""
    
    def list_volume_attachment(
        self,
        pretty: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1VolumeAttachmentList:
        """List volume attachments."""
    
    def read_volume_attachment(
        self,
        name: str,
        pretty: str = None
    ) -> V1VolumeAttachment:
        """Read specified volume attachment."""
    
    def patch_volume_attachment(
        self,
        name: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1VolumeAttachment:
        """Update specified volume attachment."""
    
    def delete_volume_attachment(
        self,
        name: str,
        dry_run: str = None,
        grace_period_seconds: int = None,
        pretty: str = None
    ) -> V1Status:
        """Delete specified volume attachment."""

CSI Drivers

Manage Container Storage Interface (CSI) driver configurations and capabilities.

class StorageV1Api:
    def create_csi_driver(
        self,
        body: V1CSIDriver,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1CSIDriver:
        """Create a CSI driver."""
    
    def list_csi_driver(
        self,
        pretty: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1CSIDriverList:
        """List CSI drivers."""
    
    def read_csi_driver(
        self,
        name: str,
        pretty: str = None
    ) -> V1CSIDriver:
        """Read specified CSI driver."""
    
    def patch_csi_driver(
        self,
        name: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1CSIDriver:
        """Update specified CSI driver."""
    
    def delete_csi_driver(
        self,
        name: str,
        dry_run: str = None,
        grace_period_seconds: int = None,
        pretty: str = None
    ) -> V1Status:
        """Delete specified CSI driver."""

CSI Node Management

Manage CSI node configurations and per-node storage capabilities.

class StorageV1Api:
    def create_csi_node(
        self,
        body: V1CSINode,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1CSINode:
        """Create a CSI node."""
    
    def list_csi_node(
        self,
        pretty: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1CSINodeList:
        """List CSI nodes."""
    
    def read_csi_node(
        self,
        name: str,
        pretty: str = None
    ) -> V1CSINode:
        """Read specified CSI node."""
    
    def patch_csi_node(
        self,
        name: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1CSINode:
        """Update specified CSI node."""
    
    def delete_csi_node(
        self,
        name: str,
        dry_run: str = None,
        grace_period_seconds: int = None,
        pretty: str = None
    ) -> V1Status:
        """Delete specified CSI node."""

CSI Storage Capacity

Manage CSI storage capacity tracking and reporting for dynamic provisioning decisions.

class StorageV1Api:
    def create_namespaced_csi_storage_capacity(
        self,
        namespace: str,
        body: V1CSIStorageCapacity,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1CSIStorageCapacity:
        """Create CSI storage capacity information."""
    
    def list_namespaced_csi_storage_capacity(
        self,
        namespace: str,
        pretty: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1CSIStorageCapacityList:
        """List CSI storage capacity in specified namespace."""
    
    def read_namespaced_csi_storage_capacity(
        self,
        name: str,
        namespace: str,
        pretty: str = None
    ) -> V1CSIStorageCapacity:
        """Read specified CSI storage capacity."""
    
    def patch_namespaced_csi_storage_capacity(
        self,
        name: str,
        namespace: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1CSIStorageCapacity:
        """Update specified CSI storage capacity."""
    
    def delete_namespaced_csi_storage_capacity(
        self,
        name: str,
        namespace: str,
        dry_run: str = None,
        grace_period_seconds: int = None,
        pretty: str = None
    ) -> V1Status:
        """Delete specified CSI storage capacity."""

Resource Types

V1StorageClass

class V1StorageClass:
    api_version: str
    kind: str
    metadata: V1ObjectMeta
    provisioner: str
    parameters: dict
    reclaim_policy: str
    volume_binding_mode: str
    allow_volume_expansion: bool
    mount_options: list[str]
    allowed_topologies: list[V1TopologySelectorTerm]

V1VolumeAttachment

class V1VolumeAttachment:
    api_version: str
    kind: str
    metadata: V1ObjectMeta
    spec: V1VolumeAttachmentSpec
    status: V1VolumeAttachmentStatus

class V1VolumeAttachmentSpec:
    attacher: str
    node_name: str
    source: V1VolumeAttachmentSource

V1CSIDriver

class V1CSIDriver:
    api_version: str
    kind: str
    metadata: V1ObjectMeta
    spec: V1CSIDriverSpec

class V1CSIDriverSpec:
    attach_required: bool
    fs_group_policy: str
    pod_info_on_mount: bool
    requires_republish: bool
    storage_capacity: bool
    token_requests: list[V1TokenRequest]
    volume_lifecycle_modes: list[str]

V1CSIStorageCapacity

class V1CSIStorageCapacity:
    api_version: str
    kind: str
    metadata: V1ObjectMeta
    node_topology: V1LabelSelector
    storage_class_name: str
    capacity: str
    maximum_volume_size: str

Usage Examples

Creating a Storage Class

from kubernetes import client, config

# Load configuration
config.load_kube_config()

# Create storage API client
storage_v1 = client.StorageV1Api()

# Define storage class for SSD storage
storage_class = client.V1StorageClass(
    api_version="storage.k8s.io/v1",
    kind="StorageClass",
    metadata=client.V1ObjectMeta(
        name="fast-ssd",
        annotations={
            "storageclass.kubernetes.io/is-default-class": "false"
        }
    ),
    provisioner="kubernetes.io/aws-ebs",
    parameters={
        "type": "gp3",
        "fsType": "ext4",
        "encrypted": "true"
    },
    reclaim_policy="Delete",
    volume_binding_mode="WaitForFirstConsumer",
    allow_volume_expansion=True
)

# Create storage class
result = storage_v1.create_storage_class(body=storage_class)
print(f"Storage class created: {result.metadata.name}")

Managing CSI Drivers

# Create CSI driver configuration
csi_driver_spec = client.V1CSIDriverSpec(
    attach_required=True,
    pod_info_on_mount=False,
    volume_lifecycle_modes=["Persistent"],
    storage_capacity=True,
    fs_group_policy="ReadWriteOnceWithFSType"
)

csi_driver = client.V1CSIDriver(
    api_version="storage.k8s.io/v1",
    kind="CSIDriver",
    metadata=client.V1ObjectMeta(name="ebs.csi.aws.com"),
    spec=csi_driver_spec
)

# Create CSI driver
result = storage_v1.create_csi_driver(body=csi_driver)
print(f"CSI driver created: {result.metadata.name}")

# List all CSI drivers
csi_drivers = storage_v1.list_csi_driver()
for driver in csi_drivers.items:
    print(f"CSI driver: {driver.metadata.name}")
    print(f"  Attach required: {driver.spec.attach_required}")
    print(f"  Storage capacity: {driver.spec.storage_capacity}")

Managing Volume Attachments

# Create volume attachment for existing volume
volume_attachment_spec = client.V1VolumeAttachmentSpec(
    attacher="ebs.csi.aws.com",
    node_name="ip-192-168-1-100.ec2.internal",
    source=client.V1VolumeAttachmentSource(
        persistent_volume_name="pv-example"
    )
)

volume_attachment = client.V1VolumeAttachment(
    api_version="storage.k8s.io/v1",
    kind="VolumeAttachment",
    metadata=client.V1ObjectMeta(name="vol-attach-example"),
    spec=volume_attachment_spec
)

# Create volume attachment
result = storage_v1.create_volume_attachment(body=volume_attachment)
print(f"Volume attachment created: {result.metadata.name}")

# List all volume attachments
attachments = storage_v1.list_volume_attachment()
for attachment in attachments.items:
    print(f"Volume attachment: {attachment.metadata.name}")
    print(f"  Node: {attachment.spec.node_name}")
    print(f"  Attacher: {attachment.spec.attacher}")

CSI Storage Capacity Management

# Create CSI storage capacity information
csi_capacity = client.V1CSIStorageCapacity(
    api_version="storage.k8s.io/v1",
    kind="CSIStorageCapacity",
    metadata=client.V1ObjectMeta(
        name="ebs-capacity-zone-a",
        namespace="kube-system"
    ),
    storage_class_name="fast-ssd",
    capacity="1000Gi",
    maximum_volume_size="500Gi",
    node_topology=client.V1LabelSelector(
        match_labels={
            "topology.ebs.csi.aws.com/zone": "us-west-2a"
        }
    )
)

# Create storage capacity
result = storage_v1.create_namespaced_csi_storage_capacity(
    namespace="kube-system",
    body=csi_capacity
)
print(f"CSI storage capacity created: {result.metadata.name}")

Import Statements

from kubernetes import client
from kubernetes.client import StorageV1Api
from kubernetes.client import V1StorageClass, V1StorageClassList
from kubernetes.client import V1VolumeAttachment, V1VolumeAttachmentSpec
from kubernetes.client import V1CSIDriver, V1CSIDriverSpec
from kubernetes.client import V1CSIStorageCapacity, V1CSIStorageCapacityList

Install with Tessl CLI