Function reset

misc/python/materialize/cli/orchestratord.py:135–186

(args: argparse.Namespace)

def reset(args: argparse.Namespace):
    environments = (
        kubectl(
            "get",
            "--all-namespaces",
            "materializes",
            '-o=jsonpath={range .items[*]}{.metadata.namespace}/{.metadata.name}{"\\n"}{end}',
            cluster=args.kind_cluster_name,
        )
        .decode()
        .splitlines()
    )
    for environment in environments:
        namespace, environment_name = environment.split("/", 1)
        env_kubectl = make_env_kubectl(args, namespace)

        mz = json.loads(
            env_kubectl(
                "get",
                "materialize",
                environment_name,
                "-o",
                "json",
            )
        )
        backend_secret_name = mz["spec"]["backendSecretName"]
        mz_system_password_secret_name = mz["spec"].get("externalLoginSecretMzSystem")

        env_kubectl("delete", "--wait=true", "materialize", environment_name)
        env_kubectl("delete", "--wait=true", "secret", backend_secret_name)
        if mz_system_password_secret_name is not None:
            env_kubectl(
                "delete", "--wait=true", "secret", mz_system_password_secret_name
            )

    try:
        subprocess.check_call(
            [
                "helm",
                "uninstall",
                "orchestratord",
                f"--namespace={args.namespace}",
            ]
        )
        kubectl(
            "delete",
            "namespace",
            args.namespace,
            cluster=args.kind_cluster_name,
        )
    except subprocess.CalledProcessError:
        pass


def environment(args: argparse.Namespace):