ArgoCD Python Client

import json
import logging
import requests
import time

from collections import namedtuple

class ArgoInteractionError(Exception):
    """Any error in interacting with ArgoCD"""
    pass

AppUrl = namedtuple("AppUrl", ["name", "url"])

class ArgoClient:
    """Simple wrapper class to serve as ArgoCD client"""

    def __init__(self, server, token):
        self.base_url = f"https://{server}/api/v1/"
        self.token = token
        self.server = server

    def request(self, func, endpoint, *args, **kwargs) -> requests.Response:
        headers = kwargs.pop("headers", dict())
        headers["Authorization"] = f"Bearer {self.token}"
        try:
            response = func(self.base_url + endpoint, *args, headers=headers, **kwargs)
            response.raise_for_status()
            return response
        except Exception as e:
            raise ArgoInteractionError(
                f"Could not complete request to ArgoCD, endpoint={endpoint}, method={func}, reason={e.response.text}"
            ) from e

    def post(self, endpoint, *args, **kwargs) -> requests.Response:
        return self.request(requests.post, endpoint, *args, **kwargs)

    def get(self, endpoint, *args, **kwargs) -> requests.Response:
        return self.request(requests.get, endpoint, *args, **kwargs)

    def put(self, endpoint, *args, **kwargs) -> requests.Response:
        return self.request(requests.put, endpoint, *args, **kwargs)

    def patch(self, endpoint, *args, **kwargs) -> requests.Response:
        return self.request(requests.patch, endpoint, *args, **kwargs)

    def delete(self, endpoint, *args, **kwargs) -> requests.Response:
        return self.request(requests.delete, endpoint, *args, **kwargs)

    def get_app(self, name) -> dict:
        """Get an app"""
        try:
            response = self.get(f"applications/{name}")
        except ArgoInteractionError:
            return False
        data = response.json()
        return data

    def get_apps(self, project) -> dict:
        """Get all apps in a given project"""
        params = {
            "project": [
                project,
            ]
        }
        response = self.get("applications", params=params)
        data = response.json()
        return data["items"]

    def delete_app(self, name) -> dict:
        """Delete an app"""
        response = self.delete(f"applications/{name}")
        data = response.json()
        return data

    def upsert_app(
        self,
        environment_id,
        project,
        target_namespace,
        target_cluster,
    ) -> dict:
        """Creates or updates (upsert) an app"""
        app_name = f"{environment_id}-{project.path_slug}"
        logging.info(f"Upserting app {app_name}")

        data = {
            "metadata": {"name": app_name},
            "spec": {
                "destination": {"namespace": target_namespace, "server": target_cluster},
                "project": "envbuilder",
                "source": {
                    "path": ".deploy",
                    "repoURL": project.git_ssh_uri,
                    "targetRevision": project.ref,
                },
                "syncPolicy": {
                    "automated": {
                        "alloEmtpy": True,
                        "prune": True,
                        # For now we can't self heal as we have shared secrets that result in a never ending heal circle
                        "selfHeal": False,
                    },
                    "syncOptions": [
                        "CreateNamespace=true",
                    ],
                },
            },
        }
        logging.debug(f"Upsert payload: {data}")

        params = {"upsert": True}
        response = self.post("applications", params=params, json=data)
        data = response.json()
        logging.debug(f"Upsert response: {data}")
        return data

    def get_app_ressources(self, name, retries=1, wait=0) -> List[dict]:
        """Gets an apps ressources, retrying with the given wait if no ressources are found"""
        remaining_retries = retries
        while remaining_retries > 0:
            logging.info(f"{name}: Trying to get app resources, {remaining_retries}")
            response = self.get(f"applications/{name}/managed-resources")
            data = response.json()
            try:
                items = data["items"]
            except KeyError:
                pass
            else:
                if len(items) > 0:
                    return items
            remaining_retries -= 1
            time.sleep(wait)
        raise ArgoInteractionError(f"Could not get application ressources for {name}")

    def get_urls_of_app(self, name) -> List[AppUrl]:
        """Get URLs of an app based on it's ingress and route objects

        On route objects the label 'url-suffix' can be set to create a suffix for the resulting URL"""
        links = list()
        ressources = self.get_app_ressources(name, retries=2, wait=10)
        for ressource in ressources:
            # When a ressource can't be fully deployed its live state is the string "null"
            # and therefore we can't parse it
            if ressource.get("liveState", "") == "null":
                continue

            if ressource["kind"].lower() == "route":
                ressource_data = json.loads(ressource["liveState"])

                # Try to find a URL suffix label
                labels = ressource_data["metadata"].get("labels", {})
                url_suffix = labels.get("url-suffix", "")

                links.append(
                    AppUrl(
                        name=ressource_data["metadata"]["name"],
                        url=f"https://{ressource_data['spec']['host']}/{url_suffix}",
                    )
                )
            elif ressource["kind"].lower() == "ingress":
                ressource_data = json.loads(ressource["liveState"])
                for rule in ressource_data["spec"]["rules"]:
                    links.append(
                        AppUrl(
                            name=ressource_data["metadata"]["name"],
                            url=f"https://{rule['host']}",
                        )
                    )
        return links