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:
$ python3 update.py

So, We are successfully done now. This is how we can play with the Kubernetes Python client
Reference:
https://kubernetes.io/docs/reference/using-api/client-libraries/