How To Update & Delete Secrets Using Kubernetes Python Client

Hello Readers !!! Hope you all are doing well . Today in this Blog we’ll see how to Update And Delete Secrets Using Kubernetes Python Client. As we all know that generally we use kubectl commands for creating, listing, updating, deleting the kubernetes resources. But In this blog we will see how we can use python for doing these things with resources. 

Introduction

A secret is a small piece of confidential data, such as a password, token, or key. Alternatively, such information might be include in a Pod specification or a container image. You don’t have to incorporate confidential data in your application code if you use a Secret.

There is less risk of the Secret (and its data) being expose during the workflow of generating, viewing, and updating Pods because Secrets can be establish independently of the Pods that use them. Secrets can also be handle differently by Kubernetes and the applications that run in your cluster, such as avoiding writing secret data to nonvolatile storage.

Secrets are similar to Config Maps, except they’re designe to retain sensitive information.

Prerequisites

Before we move forward with creating secrets using 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:

$ pip install kubernetes

Now, we have the python-kubernetes package-installed.

So, let’s get start!!!

Updating Secrets

let’s see How To Update & Delete Secrets Using Kubernetes Python Client:

My Update_Delete_Secret.py file contains the following code for Updating and Deleting a secrets using Kubernetes Python Client. Now that we have the python-kubernetes package installed, we can import it as:

from kubernetes import client, config

Here is my Code for Updating and Deleting Secrets Using Kubernetes Python Client

from kubernetes import client
from kubernetes.client import ApiClient
import json
import yaml
from kubernetes.client.rest import ApiException


def __get_kubernetes_corev1client(bearer_token,api_server_endpoint):
    try:
        configuration = client.Configuration()
        configuration.host = api_server_endpoint
        configuration.verify_ssl = False
        configuration.api_key = {"authorization": "Bearer " + bearer_token}
        client.Configuration.set_default(configuration)
        client_api = client.CoreV1Api()
        return client_api
    except Exception as e:
        print("Error getting kubernetes client \n{}".format(e))
        return None

def __format_data_for_secret(client_output):
        temp_dict={}
        temp_list=[]
        json_data=ApiClient().sanitize_for_serialization(client_output)
        if len(json_data["items"]) != 0:
            for secret in json_data["items"]:
                temp_dict={
                    "secret": secret["metadata"]["name"],
                    "namespace": secret["metadata"]["namespace"]
                }
                temp_list.append(temp_dict)
        return temp_list

def __format_data_for_create_secret(client_output):
        temp_dict={}
        temp_list=[]
        json_data=ApiClient().sanitize_for_serialization(client_output)
        
        if type(json_data) is str:
            print("FORMAT_DATA :{}".format(type(json_data)))
            json_data = json.loads(json_data)
        temp_list.append(json_data)
        return temp_list
    

def create_secret(cluster_details,yaml_body=None,namespace="default"):
    try:
        client_api= __get_kubernetes_corev1client(
                bearer_token=cluster_details["bearer_token"],
                api_server_endpoint=cluster_details["api_server_endpoint"],
            )
        yaml_data=open("secret.yaml", "rb").read().decode('utf-8')
        yaml_body=yaml.safe_load(yaml_data)
        resp = client_api.create_namespaced_secret(
            body=yaml_body, namespace="{}".format(namespace))

        data=__format_data_for_create_secret(resp)
        print (data)    
    except ApiException as e:
        print("ERROR IN create_secret:\n{}".format(e.body))
        print("TYPE :{}".format(type(e)))
        return __format_data_for_create_secret(e.body)

def update_secret(cluster_details,k8s_object_name=None,yaml_body=None,namespace="default"):
    # Configs can be set in Configuration class directly or using helper
    # utility. If no argument provided, the config will be loaded from
    # default location.
    try:
        client_api= __get_kubernetes_corev1client(
                bearer_token=cluster_details["bearer_token"],
                api_server_endpoint=cluster_details["api_server_endpoint"],
            )
        resp = client_api.patch_namespaced_secret(
            name=k8s_object_name,
            body=yaml_body, 
            namespace="{}".format(namespace))

        data=__format_data_for_create_secret(resp)
        return data
    except ApiException as e:
        print("ERROR IN create_deployment:\n{}".format(e.body))
        print("TYPE :{}".format(type(e)))
        return __format_data_for_create_secret(e.body)


def delete_secret(cluster_details,k8s_object_name=None,namespace="default"):
    # Configs can be set in Configuration class directly or using helper
    # utility. If no argument provided, the config will be loaded from
    # default location.
    try:
        client_api= __get_kubernetes_corev1client(
                bearer_token=cluster_details["bearer_token"],
                api_server_endpoint=cluster_details["api_server_endpoint"],
            )
        resp = client_api.delete_namespaced_secret(
                name=k8s_object_name,
                namespace="{}".format(namespace),
                body=client.V1DeleteOptions(
                    propagation_policy="Foreground", grace_period_seconds=5)
            )

        data=__format_data_for_create_secret(resp)
        return data
    except ApiException as e:
        print("ERROR IN create_deployment:\n{}".format(e.body))
        print("TYPE :{}".format(type(e)))
        return __format_data_for_create_secret(e.body)

if __name__ == '__main__':
    cluster_details={
        "bearer_token":"Your Bearer Token",
        "api_server_endpoint":"Your API"
    }

    #create_secret(cluster_details,"default")
    #update_secret(cluster_details,k8s_object_name="deekshaa")
    #delete_secret(cluster_details,k8s_object_name="deekshaa")

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

As you can see here my secrets named –> deekshaa is successfully created.

let’s see How to update Secrets. I have update the secret.yaml as:

I have change the data name in the secret.yaml file.

---
kind: Secret
apiVersion: v1
metadata:
  name: deekshaa
  namespace: default

data:
  name: bmFpbmN5
type: Opaque

---
kind: Secret
apiVersion: v1
metadata:
  name: deekshaa
  namespace: default

data:
  name: abcdefgh
type: Opaque

As you can see I have updated the Update_Delete_Secret.yaml:

Now let’s see How to Delete that :

Run the delete function and you’ll see

So, We are successfully done now. This is how we can play with the Kubernetes python client.

Conclusion

Thank you for sticking to the end. In this blog we have seen how easily we can Create, Update And Delete Secrets Using Kubernetes Python Client If you like this blog, please do show your appreciation by giving thumbs ups and share this blog and give me suggestions on how I can improve my future posts to suit your needs.

HAPPY LEARNING!!!

Thank You!!!