argocd-sync

#!/usr/bin/env python3

import argparse
import json
import re
import subprocess
import time
from typing import Any

import yaml


def run_retry(cmd: list[str], nb_times: int = 150, delay=10, **kwargs: Any) -> None:
    error = None
    for n in range(nb_times):
        try:
            return subprocess.run(cmd, **kwargs)
        except subprocess.CalledProcessError as err:
            print(err)
            if n < nb_times - 1:
                print(f"Will retry in {delay}s.")
                time.sleep(delay)
            error = err
    raise error


def main() -> None:
    """
    Sync the application on ArgoCD on each commit on GitHub master branch.
    """

    parser = argparse.ArgumentParser(
        description="Sync the application on ArgoCD on each commit on GitHub master branch."
    )
    parser.add_argument("--dry-run", help="Dry run.", action="store_true")
    args = parser.parse_args()

    app_list = json.loads(
        run_retry(
            ["argocd", "app", "list", "--output=json"], stdout=subprocess.PIPE, check=True
        ).stdout.decode("utf-8")
    )
    apps = {app["metadata"]["name"] for app in app_list}

    with open("data/argocd.yaml", encoding="utf-8") as no_sync_file:
        no_sync_config = yaml.load(no_sync_file, Loader=yaml.SafeLoader)
        no_sync_re = [re.compile(pattern) for pattern in no_sync_config["no_sync_apps_re"]]
        apps_re = [re.compile(pattern) for pattern in no_sync_config["apps_re"]]

    ignored_apps = set()
    for app in apps:
        for no_sync in no_sync_re:
            if no_sync.match(app):
                ignored_apps.add(app)
                break

        accept = False
        for app_re in apps_re:
            if app_re.match(app):
                accept = True
                break
        if not accept:
            ignored_apps.add(app)

    sync_apps = apps - ignored_apps

    print("Ignored applications:")
    print("\n".join(ignored_apps))
    print()
    print("Application that will be synced:")
    print("\n".join(sync_apps))
    print()

    if not args.dry_run:
        run_retry(["argocd", "app", "sync", "--async"] + list(sync_apps), check=True)


if __name__ == "__main__":
    main()