How to create Job workloads Using Kubernetes Python Client -

Table of contents

Reading Time: 3 minutes

Hello Readers! In this blog, we will see how we can create a job using the Kubernetes Python client. As we all know that we use kubectl commands for creating, listing, updating, and deleting the Kubernetes resources.

What is K8s Python Client?

To write applications using the Kubernetes Rest API, you do not need to implement the API calls and request/response types yourself. You can use a client library for the programming language you are using. Client libraries often handle common tasks such as authentication for you.

Prerequisites:

Before we move forward with creating job using the K8s python client we have some prerequisites that we need to follow.

Kubernetes library provides us modules such as client and config which we will use here. So, let’s install Kubernetes Python Client:

pip3 install kubernetes

In my case I have already installed it, that’s why it shows us already satisfied.

Now, we have the python-Kubernetes package installed.

Creating a job:

Now, I will create one directory and inside this, I will make a file named job.py.

mkdir k8s

cd k8s

touch job.py

My job.py file contains the following code for creating a job using Kubernetes Python Client. Now that we have the python-Kubernetes package installed, we can import it as:

from kubernetes import client, config

Inside this code I am creating functions first one is create_service( ) and second one is create_job_object( ) and third one is create_job( ) . I am creating this job in my default namespace, if you want you can use another also.

from os import path
from time import sleep

import yaml

from kubernetes import client, config

JOB_NAME = "pi"


def create_job_object():
    # Configureate Pod template container
    container = client.V1Container(
        name="pi",
        image="perl",
        command=["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"])
    # Create and configure a spec section
    template = client.V1PodTemplateSpec(
        metadata=client.V1ObjectMeta(labels={"app": "pi"}),
        spec=client.V1PodSpec(restart_policy="Never", containers=[container]))
    # Create the specification of deployment
    spec = client.V1JobSpec(
        template=template,
        backoff_limit=4)
    # Instantiate the job object
    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))



def main():
    config.load_kube_config()
    batch_v1 = client.BatchV1Api()
    job = create_job_object()

    create_job(batch_v1, job)


if __name__ == '__main__':
    main()

Now, It’s time to create the job. So, I will now run the python code.

$ python3 job.py

As you can see here my job is successfully created.

Updating Job:

For updating this job I will create a file named update.py. Inside this file, I will update the container image here. My file contains the following code for updating the job:

from os import path
from time import sleep

import yaml

from kubernetes import client, config

JOB_NAME = "pi"


def create_job_object():
    # Configureate Pod template container
    container = client.V1Container(
        name="pi",
        image="perl",
        command=["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"])
    # Create and configure a spec section
    template = client.V1PodTemplateSpec(
        metadata=client.V1ObjectMeta(labels={"app": "pi"}),
        spec=client.V1PodSpec(restart_policy="Never", containers=[container]))
    # Create the specification of deployment
    spec = client.V1JobSpec(
        template=template,
        backoff_limit=4)
    # Instantiate the job object
    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))
    get_job_status(api_instance)

def get_job_status(api_instance):
    job_completed = False
    while not job_completed:
        api_response = api_instance.read_namespaced_job_status(
            name=JOB_NAME,
            namespace="default")
        if api_response.status.succeeded is not None or \
                api_response.status.failed is not None:
            job_completed = True
        sleep(1)
        print("Job status='%s'" % str(api_response.status))


def update_job(api_instance, job):
    # Update container image
    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))

def main():
    config.load_kube_config()
    batch_v1 = client.BatchV1Api()
    job = create_job_object()

    create_job(batch_v1, job)

    update_job(batch_v1, job)


if __name__ == '__main__':
    main()

Here in this code, I have used the update_job( ) function for updating this job. Inside this function, I am updating the container image. Now, I will run the code for updating: