k8s python client that updates an image tag

image_updater.py

#!/usr/bin/env python3

# https://pip.wtf
def pip_wtf(command):
    import os, os.path, sys
    t = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".pip_wtf." + os.path.basename(__file__))
    sys.path = [p for p in sys.path if "-packages" not in p] + [t]
    os.environ["PATH"] += os.pathsep + t + os.path.sep + "bin"
    os.environ["PYTHONPATH"] = os.pathsep.join(sys.path)
    if os.path.exists(t): return
    os.system(" ".join([sys.executable, "-m", "pip", "install", "-t", t, command]))

# Now you just call it to install your packages:
#   pip_wtf('the rest of the pip install command here')
# Here are some examples for different platforms:

import sys
pip_wtf('kubernetes icecream halo datetime semver subprocess.run')

from kubernetes import client, config, utils
from kubernetes.client.rest import ApiException
from icecream import ic
from datetime import datetime as dt
from halo import Halo
import os
import time
import subprocess
import json
import semver

# Optional environment settings
CONTEXT = os.getenv('KUBE_CONTEXT', 'aks1-eastus-dev')
DEPLOYMENT_NAME = os.getenv('DEPLOYMENT_NAME', 'nginx-deploy')
IMAGE_NAME = os.getenv('IMAGE_NAME', 'crate/grand-central')
NS_LABEL_SELECTOR = os.getenv('LABEL_SELECTOR', 'kubernetes.io/metadata.name=default')
REGISTRY_NAME = os.getenv('REGISTRY_NAME', 'cr8cloud')

# Always use the latest available tag
LATEST_ONLY = os.getenv('LATEST_ONLY', 'false')
# Maximum number of retries before rolling back
MAX_RETRIES = 7

def check_ready(api, namespace, deployment, old_tag):
    # Wait for the deployment to be ready
    retry = 0
    while True:
        spinner.start()
        # Get the updated deployment
        deployment = api.read_namespaced_deployment(DEPLOYMENT_NAME, namespace.metadata.name)

        # Check the status of the deployment
        if deployment.status.updated_replicas == 1 and \
            deployment.status.unavailable_replicas is None:
            spinner.stop()
            ic("Tag Updated", namespace.metadata.name, actual_latest_tag)
            break

        time.sleep(1)
        retry += 1
        if retry == MAX_RETRIES:
            spinner.stop()
            ic("Rolling back",namespace.metadata.name ,old_tag)
            deployment.spec.template.spec.containers[0].image = old_tag
            api.replace_namespaced_deployment(DEPLOYMENT_NAME, namespace.metadata.name, deployment)
            break

def get_latest_tag(registry_name, image_name):
    # Call the Azure CLI to get the list of tags
    result = subprocess.check_output(
        [
            "az", "acr", "repository", "show-tags",
            "--name", registry_name,
            "--repository", image_name,
            "--orderby", "time_desc",
            "--output", "json",
        ],
        text=True,
    )

    # Parse the output as JSON
    tags = json.loads(result)

    # Filter out the tags that are not valid semver versions
    semver_tags = [tag for tag in tags if is_semver(tag)]

    # Get the latest semver tag
    if not semver_tags or LATEST_ONLY == 'true':
        latest_semver_tag = tags[0]
    else:
        latest_semver_tag = semver_tags[0]

    return latest_semver_tag

def is_semver(tag):
    try:
        semver.parse_version_info(tag)
        return True
    except ValueError:
        return False


def ts():
   return '%s |> ' % dt.now().strftime('%Y-%m-%d %H:%M:%S')

ic.configureOutput(prefix=ts)

ic(REGISTRY_NAME, IMAGE_NAME)
latest_tag = get_latest_tag(REGISTRY_NAME, IMAGE_NAME)
actual_latest_tag = IMAGE_NAME + ":" + latest_tag
ic(actual_latest_tag)

#----actual_latest_tag = os.getenv('IMAGE_WITH_TAG', 'nginx:1.14.3') -- for testing only
# switch to a specific k8s context provided by an environment variable
config.load_kube_config(context=CONTEXT)
api = client.AppsV1Api()
core_api = client.CoreV1Api()
spinner = Halo(text='Waiting for Deployment to get ready...', spinner='dots')

ic(CONTEXT)

# List all namespaces with the label corresponding NS_LABEL_SELECTOR
namespaces = core_api.list_namespace(label_selector=NS_LABEL_SELECTOR)

# For each namespace, get the specified deployment
for namespace in namespaces.items:
    try:
        # Get the existing deployment
        deployment = api.read_namespaced_deployment(DEPLOYMENT_NAME, namespace.metadata.name)
        ic(namespace.metadata.name, deployment.metadata.name, deployment.spec.template.spec.containers[0].image)

        # Update the image in the deployment spec
        old_tag = deployment.spec.template.spec.containers[0].image
        deployment.spec.template.spec.containers[0].image = actual_latest_tag

        # Apply the updated deployment
        api.replace_namespaced_deployment(DEPLOYMENT_NAME, namespace.metadata.name, deployment)

        check_ready(api, namespace, deployment, old_tag)

    except ApiException as e:
        if e.status == 404:
          #  ic(f"Deployment {DEPLOYMENT_NAME} not found in namespace {namespace.metadata.name}")
            ic("Deployment not found", DEPLOYMENT_NAME, namespace.metadata.name)
        else:
            raise