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

application-workloads.mddocs/

Application Workloads

Management of application deployment patterns including deployments, replica sets, daemon sets, stateful sets, jobs, and cron jobs. These resources provide scalable and reliable application hosting patterns with different lifecycle and scaling characteristics.

Capabilities

Deployment Management

Manage deployments for stateless applications with declarative updates, rolling deployments, and automatic rollback capabilities.

class AppsV1Api:
    def create_namespaced_deployment(
        self,
        namespace: str,
        body: V1Deployment,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1Deployment:
        """Create a deployment in specified namespace."""
    
    def list_namespaced_deployment(
        self,
        namespace: 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
    ) -> V1DeploymentList:
        """List deployments in specified namespace."""
    
    def read_namespaced_deployment(
        self,
        name: str,
        namespace: str,
        pretty: str = None
    ) -> V1Deployment:
        """Read specified deployment."""
    
    def patch_namespaced_deployment(
        self,
        name: str,
        namespace: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        force: bool = None,
        pretty: str = None
    ) -> V1Deployment:
        """Patch specified deployment."""
    
    def delete_namespaced_deployment(
        self,
        name: str,
        namespace: 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 deployment."""

StatefulSet Management

Manage stateful applications that require stable network identities, persistent storage, and ordered deployment/scaling.

class AppsV1Api:
    def create_namespaced_stateful_set(
        self,
        namespace: str,
        body: V1StatefulSet,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1StatefulSet:
        """Create a stateful set in specified namespace."""
    
    def list_namespaced_stateful_set(
        self,
        namespace: 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
    ) -> V1StatefulSetList:
        """List stateful sets in specified namespace."""
    
    def read_namespaced_stateful_set(
        self,
        name: str,
        namespace: str,
        pretty: str = None
    ) -> V1StatefulSet:
        """Read specified stateful set."""

DaemonSet Management

Manage daemon sets that ensure a copy of a pod runs on every node in the cluster or a subset of nodes.

class AppsV1Api:
    def create_namespaced_daemon_set(
        self,
        namespace: str,
        body: V1DaemonSet,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1DaemonSet:
        """Create a daemon set in specified namespace."""
    
    def list_namespaced_daemon_set(
        self,
        namespace: 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
    ) -> V1DaemonSetList:
        """List daemon sets in specified namespace."""

Job Management

Manage batch jobs that run pods to completion for one-time or batch processing tasks.

class BatchV1Api:
    def create_namespaced_job(
        self,
        namespace: str,
        body: V1Job,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1Job:
        """Create a job in specified namespace."""
    
    def list_namespaced_job(
        self,
        namespace: 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
    ) -> V1JobList:
        """List jobs in specified namespace."""
    
    def read_namespaced_job(
        self,
        name: str,
        namespace: str,
        pretty: str = None
    ) -> V1Job:
        """Read specified job."""
    
    def delete_namespaced_job(
        self,
        name: str,
        namespace: 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 job."""

CronJob Management

Manage scheduled jobs that run on a time-based schedule similar to cron.

class BatchV1Api:
    def create_namespaced_cron_job(
        self,
        namespace: str,
        body: V1CronJob,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1CronJob:
        """Create a cron job in specified namespace."""
    
    def list_namespaced_cron_job(
        self,
        namespace: 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
    ) -> V1CronJobList:
        """List cron jobs in specified namespace."""

ReplicaSet Management

Manage replica sets for maintaining a stable set of replica pods running at any given time.

class AppsV1Api:
    def create_namespaced_replica_set(
        self,
        namespace: str,
        body: V1ReplicaSet,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1ReplicaSet:
        """Create a replica set in specified namespace."""
    
    def list_namespaced_replica_set(
        self,
        namespace: 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
    ) -> V1ReplicaSetList:
        """List replica sets in specified namespace."""

Resource Models

V1Deployment

class V1Deployment:
    api_version: str  # "apps/v1"
    kind: str  # "Deployment"
    metadata: V1ObjectMeta
    spec: V1DeploymentSpec
    status: V1DeploymentStatus

V1StatefulSet

class V1StatefulSet:
    api_version: str  # "apps/v1"
    kind: str  # "StatefulSet"
    metadata: V1ObjectMeta
    spec: V1StatefulSetSpec
    status: V1StatefulSetStatus

V1DaemonSet

class V1DaemonSet:
    api_version: str  # "apps/v1"
    kind: str  # "DaemonSet"
    metadata: V1ObjectMeta
    spec: V1DaemonSetSpec
    status: V1DaemonSetStatus

V1Job

class V1Job:
    api_version: str  # "batch/v1"
    kind: str  # "Job"
    metadata: V1ObjectMeta
    spec: V1JobSpec
    status: V1JobStatus

V1CronJob

class V1CronJob:
    api_version: str  # "batch/v1"
    kind: str  # "CronJob"
    metadata: V1ObjectMeta
    spec: V1CronJobSpec
    status: V1CronJobStatus

V1ReplicaSet

class V1ReplicaSet:
    api_version: str  # "apps/v1"
    kind: str  # "ReplicaSet"
    metadata: V1ObjectMeta
    spec: V1ReplicaSetSpec
    status: V1ReplicaSetStatus

Specification Models

V1DeploymentSpec

class V1DeploymentSpec:
    replicas: int
    selector: V1LabelSelector
    template: V1PodTemplateSpec
    strategy: V1DeploymentStrategy
    min_ready_seconds: int
    revision_history_limit: int
    paused: bool
    progress_deadline_seconds: int

V1StatefulSetSpec

class V1StatefulSetSpec:
    replicas: int
    selector: V1LabelSelector
    template: V1PodTemplateSpec
    service_name: str
    volume_claim_templates: list
    update_strategy: V1StatefulSetUpdateStrategy
    pod_management_policy: str
    revision_history_limit: int

V1JobSpec

class V1JobSpec:
    template: V1PodTemplateSpec
    completions: int
    parallelism: int
    backoff_limit: int
    active_deadline_seconds: int
    ttl_seconds_after_finished: int
    completion_mode: str
    suspend: bool

V1CronJobSpec

class V1CronJobSpec:
    schedule: str  # Cron format schedule
    job_template: V1JobTemplateSpec
    starting_deadline_seconds: int
    concurrency_policy: str  # Allow, Forbid, Replace
    suspend: bool
    successful_jobs_history_limit: int
    failed_jobs_history_limit: int
    time_zone: str

Usage Examples

Creating a Deployment

from kubernetes import client, config

config.load_kube_config()
apps_v1 = client.AppsV1Api()

deployment_manifest = {
    "apiVersion": "apps/v1",
    "kind": "Deployment",
    "metadata": {
        "name": "nginx-deployment",
        "labels": {"app": "nginx"}
    },
    "spec": {
        "replicas": 3,
        "selector": {
            "matchLabels": {"app": "nginx"}
        },
        "template": {
            "metadata": {
                "labels": {"app": "nginx"}
            },
            "spec": {
                "containers": [{
                    "name": "nginx",
                    "image": "nginx:1.20",
                    "ports": [{"containerPort": 80}],
                    "resources": {
                        "requests": {"cpu": "100m", "memory": "128Mi"},
                        "limits": {"cpu": "500m", "memory": "512Mi"}
                    }
                }]
            }
        }
    }
}

# Create deployment
deployment = apps_v1.create_namespaced_deployment(
    namespace="default",
    body=deployment_manifest
)
print(f"Deployment created: {deployment.metadata.name}")

Managing a StatefulSet

from kubernetes import client, config

config.load_kube_config()
apps_v1 = client.AppsV1Api()

statefulset_manifest = {
    "apiVersion": "apps/v1",
    "kind": "StatefulSet",
    "metadata": {
        "name": "mysql-statefulset"
    },
    "spec": {
        "serviceName": "mysql",
        "replicas": 3,
        "selector": {
            "matchLabels": {"app": "mysql"}
        },
        "template": {
            "metadata": {
                "labels": {"app": "mysql"}
            },
            "spec": {
                "containers": [{
                    "name": "mysql",
                    "image": "mysql:8.0",
                    "env": [{"name": "MYSQL_ROOT_PASSWORD", "value": "secret"}],
                    "volumeMounts": [{
                        "name": "mysql-storage",
                        "mountPath": "/var/lib/mysql"
                    }]
                }]
            }
        },
        "volumeClaimTemplates": [{
            "metadata": {"name": "mysql-storage"},
            "spec": {
                "accessModes": ["ReadWriteOnce"],
                "resources": {"requests": {"storage": "10Gi"}}
            }
        }]
    }
}

# Create StatefulSet
sts = apps_v1.create_namespaced_stateful_set(
    namespace="default",
    body=statefulset_manifest
)
print(f"StatefulSet created: {sts.metadata.name}")

Creating a CronJob

from kubernetes import client, config

config.load_kube_config()
batch_v1 = client.BatchV1Api()

cronjob_manifest = {
    "apiVersion": "batch/v1",
    "kind": "CronJob",
    "metadata": {
        "name": "backup-cronjob"
    },
    "spec": {
        "schedule": "0 2 * * *",  # Run daily at 2 AM
        "jobTemplate": {
            "spec": {
                "template": {
                    "spec": {
                        "containers": [{
                            "name": "backup",
                            "image": "backup-tool:latest",
                            "command": ["backup-script.sh"]
                        }],
                        "restartPolicy": "OnFailure"
                    }
                }
            }
        },
        "concurrencyPolicy": "Forbid",
        "successfulJobsHistoryLimit": 3,
        "failedJobsHistoryLimit": 1
    }
}

# Create CronJob
cronjob = batch_v1.create_namespaced_cron_job(
    namespace="default",
    body=cronjob_manifest
)
print(f"CronJob created: {cronjob.metadata.name}")

Scaling a Deployment

from kubernetes import client, config

config.load_kube_config()
apps_v1 = client.AppsV1Api()

# Scale deployment to 5 replicas
scale_patch = {
    "spec": {
        "replicas": 5
    }
}

apps_v1.patch_namespaced_deployment(
    name="nginx-deployment",
    namespace="default",
    body=scale_patch
)
print("Deployment scaled to 5 replicas")

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