Mastering Kubernetes Jobs with Python: Practical SDK Examples and Tips

Install the official Python client for Kubernetes:

pip install kubernetes

Initialization

Load the local kubeconfig (default ~/.kube/config) and create a BatchV1Api instance to work with Jobs.

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

load_kube_config()
batch = BatchV1Api()

Creating a Job (SDK objects)

The following example mirrors the official job_crud.py and builds a V1Job object programmatically.

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, many developers prefer the readability of a YAML definition.

Creating a Job with YAML

Save a minimal Job manifest as job.yaml and load it as a Python dict:

---
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 f:
    cfg = yaml.safe_load(f)
job = batch.create_namespaced_job(namespace='default', body=cfg)
assert isinstance(job, V1Job)

The create_namespaced_job call accepts a dictionary, so the YAML can be passed directly after parsing.

Creating a Job with a Python dict

Because the dict structure mirrors the YAML, you can construct the Job definition entirely in code, which offers flexibility for dynamic modifications.

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)

Monitoring Job Execution

Instead of polling, use the Kubernetes Watch API to receive events only when the Job state changes.

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)
    # Process event['type'] (ADDED, MODIFIED, DELETED) as needed

The event type indicates whether the Job was added, modified, or deleted, and the label selector limits the stream to the specific Job.

Inspecting Job Status

After creation, the returned V1Job contains a .status field with details such as active, completion_time, conditions, and succeeded. You can also call job.to_dict() to work with a plain dictionary.

Listing Jobs

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)
    # Process each job

Omit label_selector to retrieve all Jobs in the namespace, or add a selector to filter.

Reading a Specific Job

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 but with a focus on the .status field.

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

The limit parameter can be adjusted or removed if multiple Pods are expected.

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)

Using propagation_policy='Background' ensures that the Job’s Pods are also removed; the default Orphan policy would leave them behind. 'Foreground' can be used to block until all dependent resources are deleted.

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

Without a label_selector, this call deletes every Job in the specified namespace.

Updating a Job

Updates are less common because Jobs are usually recreated, but you can patch an existing Job similarly to creation.

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

Using the Python Kubernetes SDK to manage Jobs provides a concise, programmatic alternative to raw kubectl commands, supporting creation via objects, YAML, or dicts, real‑time monitoring with Watch, and full CRUD operations.