CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-kubernetes

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

—

Pending
Overview
Eval results
Files

custom-resources.mddocs/

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 for dynamic resource management and custom operators.

Capabilities

Custom Resource Definition Management

Create and manage custom resource definitions (CRDs) that extend the Kubernetes API with new resource types.

class ApiextensionsV1Api:
    def create_custom_resource_definition(
        self,
        body: V1CustomResourceDefinition,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1CustomResourceDefinition:
        """Create a custom resource definition."""
    
    def list_custom_resource_definition(
        self,
        pretty: str = None,
        allow_watch_bookmarks: bool = None,
        continue_: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        resource_version: str = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1CustomResourceDefinitionList:
        """List custom resource definitions."""
    
    def read_custom_resource_definition(
        self,
        name: str,
        pretty: str = None
    ) -> V1CustomResourceDefinition:
        """Read specified custom resource definition."""
    
    def patch_custom_resource_definition(
        self,
        name: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        force: bool = None,
        pretty: str = None
    ) -> V1CustomResourceDefinition:
        """Patch specified custom resource definition."""
    
    def delete_custom_resource_definition(
        self,
        name: str,
        body: V1DeleteOptions = None,
        dry_run: str = None,
        grace_period_seconds: int = None,
        orphan_dependents: bool = None,
        propagation_policy: str = None,
        pretty: str = None
    ) -> V1Status:
        """Delete specified custom resource definition."""

Custom Object Operations

Manage instances of custom resources using the generic custom objects API.

class CustomObjectsApi:
    def create_namespaced_custom_object(
        self,
        group: str,
        version: str,
        namespace: str,
        plural: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> object:
        """Create a namespaced custom object."""
    
    def list_namespaced_custom_object(
        self,
        group: str,
        version: str,
        namespace: str,
        plural: str,
        pretty: str = None,
        allow_watch_bookmarks: bool = None,
        continue_: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        resource_version: str = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> object:
        """List namespaced custom objects."""
    
    def get_namespaced_custom_object(
        self,
        group: str,
        version: str,
        namespace: str,
        plural: str,
        name: str,
        pretty: str = None
    ) -> object:
        """Get a namespaced custom object."""
    
    def patch_namespaced_custom_object(
        self,
        group: str,
        version: str,
        namespace: str,
        plural: str,
        name: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        force: bool = None,
        pretty: str = None
    ) -> object:
        """Patch a namespaced custom object."""
    
    def delete_namespaced_custom_object(
        self,
        group: str,
        version: str,
        namespace: str,
        plural: str,
        name: str,
        body: V1DeleteOptions = None,
        dry_run: str = None,
        grace_period_seconds: int = None,
        orphan_dependents: bool = None,
        propagation_policy: str = None,
        pretty: str = None
    ) -> object:
        """Delete a namespaced custom object."""

Cluster-scoped Custom Objects

Manage cluster-scoped custom resources that are not namespaced.

class CustomObjectsApi:
    def create_cluster_custom_object(
        self,
        group: str,
        version: str,
        plural: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> object:
        """Create a cluster-scoped custom object."""
    
    def list_cluster_custom_object(
        self,
        group: str,
        version: str,
        plural: str,
        pretty: str = None,
        allow_watch_bookmarks: bool = None,
        continue_: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        resource_version: str = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> object:
        """List cluster-scoped custom objects."""
    
    def get_cluster_custom_object(
        self,
        group: str,
        version: str,
        plural: str,
        name: str,
        pretty: str = None
    ) -> object:
        """Get a cluster-scoped custom object."""

Resource Models

V1CustomResourceDefinition

class V1CustomResourceDefinition:
    api_version: str  # "apiextensions.k8s.io/v1"
    kind: str  # "CustomResourceDefinition"
    metadata: V1ObjectMeta
    spec: V1CustomResourceDefinitionSpec
    status: V1CustomResourceDefinitionStatus

V1CustomResourceDefinitionSpec

class V1CustomResourceDefinitionSpec:
    group: str
    names: V1CustomResourceDefinitionNames
    scope: str  # "Namespaced" or "Cluster"
    versions: list  # List of V1CustomResourceDefinitionVersion
    conversion: V1CustomResourceConversion
    preserve_unknown_fields: bool

V1CustomResourceDefinitionNames

class V1CustomResourceDefinitionNames:
    kind: str
    plural: str
    singular: str
    short_names: list
    categories: list
    list_kind: str

V1CustomResourceDefinitionVersion

class V1CustomResourceDefinitionVersion:
    name: str
    served: bool
    storage: bool
    schema: V1CustomResourceValidation
    subresources: V1CustomResourceSubresources
    additional_printer_columns: list
    deprecated: bool

Usage Examples

Creating a Custom Resource Definition

from kubernetes import client, config

config.load_kube_config()
api = client.ApiextensionsV1Api()

# Define a CRD for a custom application
crd_manifest = {
    "apiVersion": "apiextensions.k8s.io/v1",
    "kind": "CustomResourceDefinition",
    "metadata": {
        "name": "applications.mycompany.io"
    },
    "spec": {
        "group": "mycompany.io",
        "versions": [{
            "name": "v1",
            "served": True,
            "storage": True,
            "schema": {
                "openAPIV3Schema": {
                    "type": "object",
                    "properties": {
                        "spec": {
                            "type": "object",
                            "properties": {
                                "image": {
                                    "type": "string"
                                },
                                "replicas": {
                                    "type": "integer",
                                    "minimum": 1,
                                    "maximum": 10
                                },
                                "port": {
                                    "type": "integer"
                                }
                            },
                            "required": ["image", "replicas"]
                        },
                        "status": {
                            "type": "object",
                            "properties": {
                                "ready": {
                                    "type": "boolean"
                                },
                                "message": {
                                    "type": "string"
                                }
                            }
                        }
                    }
                }
            },
            "subresources": {
                "status": {}
            }
        }],
        "scope": "Namespaced",
        "names": {
            "plural": "applications",
            "singular": "application",
            "kind": "Application",
            "shortNames": ["app"]
        }
    }
}

# Create the CRD
try:
    crd = api.create_custom_resource_definition(body=crd_manifest)
    print(f"CRD created: {crd.metadata.name}")
except client.ApiException as e:
    print(f"Failed to create CRD: {e}")

Working with Custom Objects

from kubernetes import client, config

config.load_kube_config()
custom_api = client.CustomObjectsApi()

# Create a custom resource instance
application_manifest = {
    "apiVersion": "mycompany.io/v1",
    "kind": "Application",
    "metadata": {
        "name": "my-web-app",
        "namespace": "default"
    },
    "spec": {
        "image": "nginx:1.20",
        "replicas": 3,
        "port": 80
    }
}

# Create the custom resource
try:
    app = custom_api.create_namespaced_custom_object(
        group="mycompany.io",
        version="v1",
        namespace="default",
        plural="applications",
        body=application_manifest
    )
    print(f"Application created: {app['metadata']['name']}")
except client.ApiException as e:
    print(f"Failed to create application: {e}")

# List all applications
apps = custom_api.list_namespaced_custom_object(
    group="mycompany.io",
    version="v1",
    namespace="default",
    plural="applications"
)

print("Applications:")
for app in apps['items']:
    name = app['metadata']['name']
    replicas = app['spec']['replicas']
    image = app['spec']['image']
    print(f"  - {name}: {replicas} replicas of {image}")

Using Dynamic Client with Custom Resources

from kubernetes import client, config, dynamic

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

# Get the custom resource definition
applications = dyn_client.resources.get(
    api_version="mycompany.io/v1",
    kind="Application"
)

# Create instance using dynamic client
app_spec = {
    "apiVersion": "mycompany.io/v1",
    "kind": "Application",
    "metadata": {
        "name": "dynamic-app",
        "namespace": "default"
    },
    "spec": {
        "image": "httpd:2.4",
        "replicas": 2,
        "port": 8080
    }
}

# Create application
app = applications.create(body=app_spec, namespace="default")
print(f"Created application: {app.metadata.name}")

# Get specific application
app = applications.get(name="dynamic-app", namespace="default")
print(f"Application status: {getattr(app, 'status', 'No status')}")

# Update application
patch_data = {
    "spec": {
        "replicas": 5
    }
}

patched_app = applications.patch(
    name="dynamic-app",
    namespace="default",
    body=patch_data
)
print(f"Scaled application to {patched_app.spec.replicas} replicas")

Managing CRD Versions

from kubernetes import client, config

config.load_kube_config()
api = client.ApiextensionsV1Api()

def add_crd_version(crd_name, new_version_spec):
    """Add a new version to an existing CRD."""
    
    # Get existing CRD
    crd = api.read_custom_resource_definition(name=crd_name)
    
    # Add new version
    crd.spec.versions.append(new_version_spec)
    
    # Update CRD
    updated_crd = api.replace_custom_resource_definition(
        name=crd_name,
        body=crd
    )
    
    return updated_crd

# Define new version
v2_spec = {
    "name": "v2",
    "served": True,
    "storage": False,  # Keep v1 as storage version
    "schema": {
        "openAPIV3Schema": {
            "type": "object",
            "properties": {
                "spec": {
                    "type": "object",
                    "properties": {
                        "image": {"type": "string"},
                        "replicas": {
                            "type": "integer",
                            "minimum": 1,
                            "maximum": 100  # Increased limit in v2
                        },
                        "port": {"type": "integer"},
                        "healthCheck": {  # New field in v2
                            "type": "object",
                            "properties": {
                                "path": {"type": "string"},
                                "port": {"type": "integer"}
                            }
                        }
                    },
                    "required": ["image", "replicas"]
                }
            }
        }
    }
}

# Add version to CRD
try:
    updated_crd = add_crd_version("applications.mycompany.io", v2_spec)
    print(f"Added v2 to CRD. Versions: {[v['name'] for v in updated_crd.spec.versions]}")
except client.ApiException as e:
    print(f"Failed to update CRD: {e}")

Custom Resource Status Updates

from kubernetes import client, config

config.load_kube_config()
custom_api = client.CustomObjectsApi()

def update_application_status(name, namespace, status_data):
    """Update the status subresource of a custom application."""
    
    try:
        # Update status subresource
        result = custom_api.patch_namespaced_custom_object_status(
            group="mycompany.io",
            version="v1",
            namespace=namespace,
            plural="applications",
            name=name,
            body={"status": status_data}
        )
        return result
    except client.ApiException as e:
        print(f"Failed to update status: {e}")
        return None

# Update application status
status_update = {
    "ready": True,
    "message": "Application is healthy and running",
    "lastUpdated": "2023-01-01T12:00:00Z",
    "readyReplicas": 3
}

updated_app = update_application_status(
    "my-web-app",
    "default",
    status_update
)

if updated_app:
    print(f"Status updated for application: {updated_app['metadata']['name']}")

Watching Custom Resources

from kubernetes import client, config, watch

config.load_kube_config()
custom_api = client.CustomObjectsApi()
w = watch.Watch()

# Watch custom resource events
print("Watching application events...")
for event in w.stream(
    custom_api.list_namespaced_custom_object,
    group="mycompany.io",
    version="v1",
    namespace="default",
    plural="applications"
):
    event_type = event['type']
    app = event['object']
    app_name = app['metadata']['name']
    
    print(f"{event_type}: Application {app_name}")
    
    if event_type in ['ADDED', 'MODIFIED']:
        spec = app.get('spec', {})
        status = app.get('status', {})
        
        print(f"  Image: {spec.get('image', 'unknown')}")
        print(f"  Replicas: {spec.get('replicas', 0)}")
        print(f"  Ready: {status.get('ready', False)}")

Cluster-scoped Custom Resources

from kubernetes import client, config

config.load_kube_config()
api = client.ApiextensionsV1Api()
custom_api = client.CustomObjectsApi()

# Create cluster-scoped CRD
cluster_crd = {
    "apiVersion": "apiextensions.k8s.io/v1",
    "kind": "CustomResourceDefinition",
    "metadata": {
        "name": "clusters.infrastructure.mycompany.io"
    },
    "spec": {
        "group": "infrastructure.mycompany.io",
        "versions": [{
            "name": "v1",
            "served": True,
            "storage": True,
            "schema": {
                "openAPIV3Schema": {
                    "type": "object",
                    "properties": {
                        "spec": {
                            "type": "object",
                            "properties": {
                                "region": {"type": "string"},
                                "nodeCount": {"type": "integer"},
                                "version": {"type": "string"}
                            }
                        }
                    }
                }
            }
        }],
        "scope": "Cluster",  # Cluster-scoped
        "names": {
            "plural": "clusters",
            "singular": "cluster",
            "kind": "Cluster"
        }
    }
}

# Create cluster-scoped CRD
api.create_custom_resource_definition(body=cluster_crd)

# Create cluster-scoped custom resource instance
cluster_instance = {
    "apiVersion": "infrastructure.mycompany.io/v1",
    "kind": "Cluster",
    "metadata": {
        "name": "production-cluster"
    },
    "spec": {
        "region": "us-west-2",
        "nodeCount": 5,
        "version": "1.25.0"
    }
}

# Create cluster resource (no namespace parameter)
cluster = custom_api.create_cluster_custom_object(
    group="infrastructure.mycompany.io",
    version="v1",
    plural="clusters",
    body=cluster_instance
)

print(f"Created cluster: {cluster['metadata']['name']}")

# List cluster resources
clusters = custom_api.list_cluster_custom_object(
    group="infrastructure.mycompany.io",
    version="v1",
    plural="clusters"
)

for cluster in clusters['items']:
    name = cluster['metadata']['name']
    region = cluster['spec']['region']
    nodes = cluster['spec']['nodeCount']
    print(f"Cluster {name}: {nodes} nodes in {region}")

Install with Tessl CLI

npx tessl i tessl/pypi-kubernetes

docs

application-workloads.md

autoscaling.md

configuration.md

core-resources.md

custom-resources.md

dynamic-client.md

index.md

leader-election.md

networking.md

rbac-security.md

resource-watching.md

storage.md

streaming-operations.md

utilities.md

tile.json