Managing Kubernetes Jobs with the Python SDK: Examples and Best Practices

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

<code>pip install kubernetes</code>

Initialization

<code>from kubernetes.client import BatchV1Api
from kubernetes.config import load_kube_config

load_kube_config()
batch = BatchV1Api()</code>

load_kube_config loads the configuration from the default location ~/.kube/config ; a custom path can be supplied as the first argument.

BatchV1Api() serves as the client for Job resources; the term "Batch" emphasizes the batch‑processing nature of Jobs.

Create Job

The following code is taken from the official job_crud.py example.

<code>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))
</code>

While this low‑level approach works, using a YAML manifest is often more readable.

Direct YAML Usage

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

Load the YAML file as a dictionary and submit it directly:

<code>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)
</code>

The create_namespaced_job method accepts a dictionary, so the YAML can be read and passed without conversion.

Using a Dictionary Directly

<code>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)
</code>

Because a dict mirrors the YAML structure without class constraints, it offers flexible manipulation before submission.

Monitoring Job Execution

After creating a Job, you can watch its lifecycle events using the Watch mechanism, which streams only when the Job state changes.

<code>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)
</code>

The event type can be ADDED (creation), MODIFIED (status change), or DELETED (removal); this pattern applies to other resources such as Pods and Deployments as well.

Working with V1Job Objects

Most fields of a V1Job instance reflect the original specification, with additional cluster‑generated information accessible via the .status attribute.

<code>>> 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': [{...}],
 'failed': None,
 'start_time': datetime.datetime(2020, 8, 10, 9, 49, 32, tzinfo=tzutc()),
 'succeeded': 1}
</code>

Accessing job.status.succeeded yields the number of successful containers; job.to_dict() can convert the object to a plain dict if needed.

Listing Jobs

<code>from kubernetes.client import V1JobList, V1Job

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

Omit label_selector to retrieve all Jobs in a Namespace; you can add a selector to filter specific groups.

Reading a Specific Job

<code>from kubernetes.client import V1Job

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

For status‑only information, use read_namespaced_job_status , which returns the same V1Job structure.

Listing Pods Belonging to a Job

<code>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
</code>

The function returns the Pods associated with the given Job name; the limit=1 optimization assumes a single Pod per Job.

Deleting a Job

<code>from kubernetes.client import V1Status

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

Using propagation_policy='Background' ensures that Pods are also removed; the default Orphan leaves Pods behind, which differs from kubectl 's default behavior.

To delete multiple Jobs, optionally filter with a label selector:

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

Updating a Job

<code>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))
</code>

Updates are less common because creating a new Job is usually simpler; the patch operation follows the same pattern as creation.

Conclusion

Operating Kubernetes Jobs with Python is generally straightforward, though some API quirks (such as propagation policies) require attention.