Master Python Kubernetes Jobs: Practical SDK Examples & Tips

Introduction

This article provides additional example code and explanations for operating Kubernetes Jobs with the Python SDK, beyond the basic examples in the official repository.

Installation

pip install kubernetes

Initialization

from kubernetes.client import BatchV1Api
from kubernetes.config import load_kube_config

load_kube_config()  # loads from ~/.kube/config by default; pass a path to use a different file
batch = BatchV1Api()  # client for Job resources

Creating a Job (Python objects)

def create_job_object():
    container = client.V1Container(
        name="pi",
        image="perl",
        command=["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"]
    )
    template = client.V1PodTemplateSpec(
        metadata=client.V1ObjectMeta(labels={"app": "pi"}),
        spec=client.V1PodSpec(restart_policy="Never", containers=[container])
    )
    spec = client.V1JobSpec(
        template=template,
        backoff_limit=4
    )
    job = client.V1Job(
        api_version="batch/v1",
        kind="Job",
        metadata=client.V1ObjectMeta(name=JOB_NAME),
        spec=spec
    )
    return job

def create_job(api_instance, job):
    api_response = api_instance.create_namespaced_job(
        body=job,
        namespace="default"
    )
    print("Job created. status='%s'" % str(api_response.status))

While this works, using a YAML manifest is often clearer and more maintainable.

Directly using YAML

---
apiVersion: batch/v1
kind: Job
metadata:
  name: hello
spec:
  template:
    spec:
      containers:
        - name: echo
          image: alpine:3.11
          args:
            - 'echo'
            - 'Hello world!'

from kubernetes.client import V1Job
import yaml

with open('job.yaml') as file:
    cfg = yaml.safe_load(file)
job = batch.create_namespaced_job(namespace='default', body=cfg)
assert isinstance(job, V1Job)

Using a dict (equivalent to YAML)

cfg = {
    'apiVersion': 'batch/v1',
    'kind': 'Job',
    'metadata': {'name': 'hello'},
    'spec': {
        'template': {
            'spec': {
                'restartPolicy': 'Never',
                'containers': [{
                    'name': 'upload',
                    'image': 'alpine:3.11',
                    'args': ['echo', 'Hello world!']
                }]
            }
        }
    }
}
batch.create_namespaced_job(namespace='default', body=cfg)

Because the dict structure mirrors YAML, it offers the same flexibility without the constraints of class objects.

Monitoring Job execution

from kubernetes.client import V1Job
from kubernetes.watch import Watch

job_name = 'hello'
watcher = Watch()
for event in watcher.stream(
        batch.list_namespaced_job,
        namespace='default',
        label_selector=f'job-name={job_name}',
):
    assert isinstance(event, dict)
    job = event['object']
    assert isinstance(job, V1Job)

The Watch stream provides events only when the Job state changes, making it an efficient callback mechanism. Event types include ADDED, MODIFIED, and DELETED.

Inspecting a V1Job

>> from kubernetes.client import V1JobStatus
>>> isinstance(job.status, V1JobStatus)
True
>>> print(job.status)
{'active': None,
 'completion_time': datetime.datetime(2020, 8, 10, 9, 49, 38, tzinfo=tzutc()),
 'conditions': [{'last_probe_time': datetime.datetime(2020, 8, 10, 9, 49, 38, tzinfo=tzutc()),
                 'last_transition_time': datetime.datetime(2020, 8, 10, 9, 49, 38, tzinfo=tzutc()),
                 'message': None,
                 'reason': None,
                 'status': 'True',
                 'type': 'Complete'}],
 'failed': None,
 'start_time': datetime.datetime(2020, 8, 10, 9, 49, 32, tzinfo=tzutc()),
 'succeeded': 1}

Typical usage is job.status.succeeded to get the number of successful containers.

Listing Jobs

from kubernetes.client import V1JobList, V1Job

job_list = batch.list_namespaced_job(namespace='default')
assert isinstance(job_list, V1JobList)
assert isinstance(job_list.items, list)
for job in job_list.items:
    assert isinstance(job, V1Job)

Reading a specific Job

from kubernetes.client import V1Job

job = batch.read_namespaced_job(name='hello', namespace='default')
assert isinstance(job, V1Job)

For status‑only information, use read_namespaced_job_status which returns the same V1Job object with only status fields populated.

Listing Pods belonging to a Job

from typing import List
from kubernetes.client import CoreV1Api, V1Pod

def get_pods_by(job_name: str) -> List[V1Pod]:
    core = CoreV1Api()
    pods = core.list_namespaced_pod(
        namespace='default',
        label_selector=f'job-name={job_name}',
        limit=1,
    )
    return pods.items

Deleting a Job

from kubernetes.client import V1Status

status = batch.delete_namespaced_job(
    namespace='default',
    name=job_name,
    propagation_policy='Background',
)
assert isinstance(status, V1Status)

The propagation_policy='Background' ensures that Pods are also removed; the default Orphan leaves them behind.

status = batch.delete_collection_namespaced_job(
    namespace='default',
    propagation_policy='Background',
    label_selector='some-label=your-value',
)
assert isinstance(status, V1Status)

Omitting label_selector deletes all Jobs in the specified Namespace.

Updating a Job

def update_job(api_instance, job):
    job.spec.template.spec.containers[0].image = "perl"
    api_response = api_instance.patch_namespaced_job(
        name=JOB_NAME,
        namespace="default",
        body=job)
    print("Job updated. status='%s'" % str(api_response.status))

Conclusion

Operating Kubernetes Jobs via the Python client is generally straightforward, though some API quirks—such as the need to specify a propagation policy when deleting—require attention.