K8s python client trials (#115)

.github/workflows/build.yml

@@ -35,6 +35,14 @@ jobs:
     - name: Run pre-commit
       uses: pre-commit/action@v3.0.0
 
+    - uses: actions/checkout@v4
+    - name: Retrieve the kubeconfig and decode it to a file
+      env:
+        KUBECONFIG: ${{ secrets.KUBECONFIG }}
+      run: |
+        mkdir ~/.kube
+        echo "$KUBECONFIG" > ~/.kube/config
+
     - name: Test
       run: |
         kfk

Makefile

@@ -15,7 +15,7 @@ lint:
 	python -m flake8
 
 test:
-	python -m pytest -x
+	python -m pytest
 
 build: clean
 	python -m build; twine check --strict dist/*

kfk/commands/acls.py

@@ -4,7 +4,7 @@
 
 from kfk.commands import users
 from kfk.commands.main import kfk
-from kfk.commons import print_missing_options_for_command
+from kfk.commons import raise_exception_for_missing_options
 from kfk.constants import COLON
 from kfk.kubectl_command_builder import Kubectl
 from kfk.option_extensions import NotRequiredIf
@@ -124,7 +124,7 @@ def acls(
             namespace,
         )
     else:
-        print_missing_options_for_command("acls")
+        raise_exception_for_missing_options("acls")
 
 
 def add_or_remove(

kfk/commands/clusters.py

@@ -11,16 +11,24 @@
     delete_resource_config,
     get_resource_as_stream,
     open_file_in_system_editor,
-    print_missing_options_for_command,
+    raise_exception_for_missing_options,
 )
 from kfk.config import STRIMZI_PATH, STRIMZI_VERSION
 from kfk.kubectl_command_builder import Kubectl
+from kfk.kubernetes_commons import create_using_yaml, delete_using_yaml
 from kfk.messages import Messages
 from kfk.option_extensions import NotRequiredIf
 
 
 @click.option("-y", "--yes", "is_yes", help='"Yes" confirmation', is_flag=True)
-@click.option("-n", "--namespace", help="Namespace to use")
+@click.option(
+    "-n",
+    "--namespace",
+    help="Namespace to use",
+    required=True,
+    cls=NotRequiredIf,
+    options=["is_list"],
+)
 @click.option(
     "--delete-config",
     help="A cluster configuration override to be removed for an existing cluster",
@@ -97,7 +105,7 @@ def clusters(
     elif is_alter:
         alter(cluster, replicas, zk_replicas, config, delete_config, namespace)
     else:
-        print_missing_options_for_command("clusters")
+        raise_exception_for_missing_options("clusters")
 
 
 def list(namespace):
@@ -121,7 +129,6 @@ def create(cluster, replicas, zk_replicas, config, namespace, is_yes):
         _add_config_if_provided(config, cluster_dict)
 
         cluster_yaml = yaml.dump(cluster_dict)
-
         cluster_temp_file = create_temp_file(cluster_yaml)
 
         if is_yes:
@@ -130,14 +137,8 @@ def create(cluster, replicas, zk_replicas, config, namespace, is_yes):
             open_file_in_system_editor(cluster_temp_file.name)
             is_confirmed = click.confirm(Messages.CLUSTER_CREATE_CONFIRMATION)
         if is_confirmed:
-            os.system(
-                Kubectl()
-                .create()
-                .from_file("{cluster_temp_file_path}")
-                .namespace(namespace)
-                .build()
-                .format(cluster_temp_file_path=cluster_temp_file.name)
-            )
+            create_using_yaml(cluster_temp_file.name, namespace)
+
         cluster_temp_file.close()
 
 
@@ -156,7 +157,23 @@ def delete(cluster, namespace, is_yes):
     else:
         is_confirmed = click.confirm(Messages.CLUSTER_DELETE_CONFIRMATION)
     if is_confirmed:
-        os.system(Kubectl().delete().kafkas(cluster).namespace(namespace).build())
+        with open(
+            "{strimzi_path}/examples/kafka/kafka-ephemeral.yaml".format(
+                strimzi_path=STRIMZI_PATH
+            ).format(version=STRIMZI_VERSION)
+        ) as file:
+            stream = file.read()
+
+            cluster_dict = yaml.full_load(stream)
+
+            cluster_dict["metadata"]["name"] = cluster
+
+            cluster_yaml = yaml.dump(cluster_dict)
+            cluster_temp_file = create_temp_file(cluster_yaml)
+
+            delete_using_yaml(cluster_temp_file.name, namespace)
+
+            cluster_temp_file.close()
 
 
 def alter(cluster, replicas, zk_replicas, config, delete_config, namespace):

kfk/commands/configs.py

@@ -9,7 +9,7 @@
     get_config_list,
     get_list_by_split_string,
     get_resource_as_stream,
-    print_missing_options_for_command,
+    raise_exception_for_missing_options,
 )
 from kfk.constants import COMMON_NAME_PREFIX, KAFKA_PORT, NEW_LINE, SPACE, SpecialTexts
 from kfk.kubectl_command_builder import Kubectl
@@ -121,7 +121,7 @@ def configs(
                 entity_name, None, None, add_config_list, delete_config_list, namespace
             )
     else:
-        print_missing_options_for_command("configs")
+        raise_exception_for_missing_options("configs")
 
 
 def _describe_natively(entity_type, entity_name, cluster, namespace):

kfk/commands/connect/clusters.py

@@ -13,7 +13,7 @@
     get_properties_from_file,
     get_resource_as_stream,
     open_file_in_system_editor,
-    print_missing_options_for_command,
+    raise_exception_for_missing_options,
 )
 from kfk.config import STRIMZI_PATH, STRIMZI_VERSION
 from kfk.constants import (
@@ -26,6 +26,12 @@
     SpecialTexts,
 )
 from kfk.kubectl_command_builder import Kubectl
+from kfk.kubernetes_commons import (
+    create_registry_secret,
+    create_using_yaml,
+    delete_object,
+    delete_using_yaml,
+)
 from kfk.messages import Errors, Messages
 from kfk.utils import is_valid_url
 
@@ -129,7 +135,7 @@ def clusters(
     elif is_alter:
         alter(cluster, replicas, config_file, namespace)
     else:
-        print_missing_options_for_command("connect")
+        raise_exception_for_missing_options("connect")
 
 
 def list(namespace):
@@ -223,8 +229,6 @@ def create(
             is_confirmed = click.confirm(Messages.CLUSTER_CREATE_CONFIRMATION)
 
         if is_confirmed:
-            return_code = 0
-
             if connect_properties.get(SpecialTexts.CONNECT_PLUGIN_URL) is not None:
                 username = (
                     registry_username
@@ -239,37 +243,17 @@ def create(
                     else click.prompt(Messages.IMAGE_REGISTRY_PASSWORD, hide_input=True)
                 )
 
-                return_code = os.system(
-                    Kubectl()
-                    .create()
-                    .secret(
-                        "docker-registry",
-                        f"{cluster}-push-secret",
-                        "--docker-username={username}",
-                        "--docker-password={password}",
-                        "--docker-server={server}",
-                    )
-                    .namespace(namespace)
-                    .build()
-                    .format(
-                        username=username,
-                        password=password,
-                        server=connect_properties.get(SpecialTexts.CONNECT_IMAGE).data,
-                    )
+                create_registry_secret(
+                    f"{cluster}-push-secret",
+                    connect_properties.get(SpecialTexts.CONNECT_IMAGE).data,
+                    username,
+                    password,
                 )
 
-            if return_code == 0:
-                return_code = os.system(
-                    Kubectl()
-                    .create()
-                    .from_file("{cluster_temp_file_path}")
-                    .namespace(namespace)
-                    .build()
-                    .format(cluster_temp_file_path=cluster_temp_file.name)
-                )
-                if return_code == 0:
-                    for connector_config_file in connector_config_files:
-                        connectors.create(connector_config_file, cluster, namespace)
+            create_using_yaml(cluster_temp_file.name, namespace)
+
+            for connector_config_file in connector_config_files:
+                connectors.create(connector_config_file, cluster, namespace)
 
         cluster_temp_file.close()
 
@@ -296,18 +280,23 @@ def delete(cluster, namespace, is_yes):
     else:
         is_confirmed = click.confirm(Messages.CLUSTER_DELETE_CONFIRMATION)
     if is_confirmed:
-        return_code = os.system(
-            Kubectl().delete().kafkaconnects(cluster).namespace(namespace).build()
-        )
+        with open(
+            "{strimzi_path}/examples/connect/kafka-connect.yaml".format(
+                strimzi_path=STRIMZI_PATH
+            ).format(version=STRIMZI_VERSION)
+        ) as file:
+            cluster_dict = yaml.full_load(file)
 
-        if return_code == 0:
-            os.system(
-                Kubectl()
-                .delete()
-                .secret(f"{cluster}-push-secret")
-                .namespace(namespace)
-                .build()
-            )
+            cluster_dict["metadata"]["name"] = cluster
+
+            cluster_yaml = yaml.dump(cluster_dict)
+            cluster_temp_file = create_temp_file(cluster_yaml)
+
+            delete_using_yaml(cluster_temp_file.name, namespace)
+
+            cluster_temp_file.close()
+
+        delete_object(f"{cluster}-push-secret", "secret", namespace)
 
 
 def alter(cluster, replicas, config_file, namespace):

kfk/commands/connect/connectors.py

@@ -11,11 +11,12 @@
     delete_last_applied_configuration,
     get_properties_from_file,
     get_resource_as_stream,
-    print_missing_options_for_command,
+    raise_exception_for_missing_options,
 )
 from kfk.config import STRIMZI_PATH, STRIMZI_VERSION
 from kfk.constants import SpecialTexts
 from kfk.kubectl_command_builder import Kubectl
+from kfk.kubernetes_commons import create_using_yaml, delete_using_yaml
 
 CONNECTOR_SKIPPED_PROPERTIES = (
     SpecialTexts.CONNECTOR_NAME,
@@ -78,11 +79,11 @@ def connectors(
     elif is_describe:
         describe(connector, output, namespace)
     elif is_delete:
-        delete(connector, namespace)
+        delete(connector, cluster, namespace)
     elif is_alter:
         alter(config_file, cluster, namespace)
     else:
-        print_missing_options_for_command("connectors")
+        raise_exception_for_missing_options("connectors")
 
 
 def list(cluster, namespace):
@@ -129,14 +130,7 @@ def create(config_file, cluster, namespace):
         connector_yaml = yaml.dump(connector_dict)
         connector_temp_file = create_temp_file(connector_yaml)
 
-        os.system(
-            Kubectl()
-            .create()
-            .from_file("{connector_temp_file_path}")
-            .namespace(namespace)
-            .build()
-            .format(connector_temp_file_path=connector_temp_file.name)
-        )
+        create_using_yaml(connector_temp_file.name, namespace)
 
         connector_temp_file.close()
 
@@ -157,10 +151,23 @@ def describe(connector, output, namespace):
         )
 
 
-def delete(connector, namespace):
-    os.system(
-        Kubectl().delete().kafkaconnectors(connector).namespace(namespace).build()
-    )
+def delete(connector, cluster, namespace):
+    with open(
+        "{strimzi_path}/examples/connect/source-connector.yaml".format(
+            strimzi_path=STRIMZI_PATH
+        ).format(version=STRIMZI_VERSION)
+    ) as file:
+        connector_dict = yaml.full_load(file)
+
+        connector_dict["metadata"]["name"] = connector
+        connector_dict["metadata"]["labels"]["strimzi.io/cluster"] = cluster
+
+        connector_yaml = yaml.dump(connector_dict)
+        connector_temp_file = create_temp_file(connector_yaml)
+
+        delete_using_yaml(connector_temp_file.name, namespace)
+
+        connector_temp_file.close()
 
 
 def alter(config_file, cluster, namespace):