Handling of no services case

src/edge_containers_cli/cmds/argo_commands.py

@@ -12,7 +12,12 @@
 import polars
 from ruamel.yaml import YAML
 
-from edge_containers_cli.cmds.commands import CommandError, Commands, ServicesDataFrame
+from edge_containers_cli.cmds.commands import (
+    CommandError,
+    Commands,
+    ServicesDataFrame,
+    ServicesSchema,
+)
 from edge_containers_cli.definitions import ECContext
 from edge_containers_cli.git import set_values
 from edge_containers_cli.globals import TIME_FORMAT
@@ -139,45 +144,47 @@ def _get_services(self, running_only) -> ServicesDataFrame:
         )
         app_dicts = YAML(typ="safe").load(app_resp)
 
-        if not app_dicts:
-            raise CommandError(f"No ec-services found in {self.target}")
-        for app in app_dicts:
-            resources_dict = app["status"]["resources"]
-
-            for resource in resources_dict:
-                is_ready = False
-                if resource["kind"] == "StatefulSet":
-                    name = app["metadata"]["name"]
-
-                    # check if replicas ready
-                    mani_resp = shell.run_command(
-                        f"argocd app manifests {namespace}/{name} --source live",
-                    )
-                    for resource_manifest in mani_resp.split("---")[1:]:
-                        manifest = YAML(typ="safe").load(resource_manifest)
-                        if not manifest:
-                            continue
-                        kind = manifest["kind"]
-                        resource_name = manifest["metadata"]["name"]
-                        if kind == "StatefulSet" and resource_name == name:
-                            try:
-                                is_ready = bool(manifest["status"]["readyReplicas"])
-                            except (KeyError, TypeError):  # Not ready if doesnt exist
-                                is_ready = False
-                            time_stamp = datetime.strptime(
-                                manifest["metadata"]["creationTimestamp"],
-                                "%Y-%m-%dT%H:%M:%SZ",
-                            )
-                            service_data["name"].append(name)
-                            service_data["version"].append(
-                                app["spec"]["source"]["targetRevision"]
-                            )
-                            service_data["ready"].append(is_ready)
-                            service_data["deployed"].append(
-                                datetime.strftime(time_stamp, TIME_FORMAT)
-                            )
-
-        services_df = polars.from_dict(service_data)
+        if app_dicts:
+            for app in app_dicts:
+                resources_dict = app["status"]["resources"]
+
+                for resource in resources_dict:
+                    is_ready = False
+                    if resource["kind"] == "StatefulSet":
+                        name = app["metadata"]["name"]
+
+                        # check if replicas ready
+                        mani_resp = shell.run_command(
+                            f"argocd app manifests {namespace}/{name} --source live",
+                        )
+                        for resource_manifest in mani_resp.split("---")[1:]:
+                            manifest = YAML(typ="safe").load(resource_manifest)
+                            if not manifest:
+                                continue
+                            kind = manifest["kind"]
+                            resource_name = manifest["metadata"]["name"]
+                            if kind == "StatefulSet" and resource_name == name:
+                                try:
+                                    is_ready = bool(manifest["status"]["readyReplicas"])
+                                except (
+                                    KeyError,
+                                    TypeError,
+                                ):  # Not ready if doesnt exist
+                                    is_ready = False
+                                time_stamp = datetime.strptime(
+                                    manifest["metadata"]["creationTimestamp"],
+                                    "%Y-%m-%dT%H:%M:%SZ",
+                                )
+                                service_data["name"].append(name)
+                                service_data["version"].append(
+                                    app["spec"]["source"]["targetRevision"]
+                                )
+                                service_data["ready"].append(is_ready)
+                                service_data["deployed"].append(
+                                    datetime.strftime(time_stamp, TIME_FORMAT)
+                                )
+
+        services_df = polars.from_dict(service_data, schema=ServicesSchema)
 
         if running_only:
             services_df = services_df.filter(polars.col("ready").eq(True))

src/edge_containers_cli/cmds/k8s_commands.py

@@ -124,34 +124,46 @@ def _get_services(self, running_only):
             f'kubectl get statefulset -l "is_ioc==true" -n {self.target} -o yaml',
         )
         sts_dicts = YAML(typ="safe").load(kubectl_res)
-        if not sts_dicts["items"]:
-            raise CommandError(f"No ec-services found in {self.target}")
         service_data = {
             "name": [],  # type: ignore
             "ready": [],
             "deployed": [],
         }
-        for sts in sts_dicts["items"]:
-            name = sts["metadata"]["name"]
-            time_stamp = datetime.strptime(
-                sts["metadata"]["creationTimestamp"], "%Y-%m-%dT%H:%M:%SZ"
-            )
-            try:
-                is_ready = bool(sts["status"]["readyReplicas"])
-            except KeyError:  # Not ready if doesnt exist
-                is_ready = False
-
-            # Fill app data
-            service_data["name"].append(name)
-            service_data["ready"].append(is_ready)
-            service_data["deployed"].append(datetime.strftime(time_stamp, TIME_FORMAT))
-
-        services_df = polars.from_dict(service_data)
+        if sts_dicts["items"]:
+            for sts in sts_dicts["items"]:
+                name = sts["metadata"]["name"]
+                time_stamp = datetime.strptime(
+                    sts["metadata"]["creationTimestamp"], "%Y-%m-%dT%H:%M:%SZ"
+                )
+                try:
+                    is_ready = bool(sts["status"]["readyReplicas"])
+                except KeyError:  # Not ready if doesnt exist
+                    is_ready = False
+
+                # Fill app data
+                service_data["name"].append(name)
+                service_data["ready"].append(is_ready)
+                service_data["deployed"].append(
+                    datetime.strftime(time_stamp, TIME_FORMAT)
+                )
+
+        services_df = polars.from_dict(
+            service_data,
+            schema=polars.Schema(
+                {
+                    "name": polars.String,
+                    "ready": polars.Boolean,
+                    "deployed": polars.String,
+                }
+            ),
+        )
 
         # Adds the version for all services
         helm_out = str(shell.run_command(f"helm list -n {self.target} -o json"))
         if helm_out == "[]\n":
-            helm_df = polars.DataFrame({"name": [""], "version": [""]})
+            helm_df = polars.DataFrame(
+                schema=polars.Schema({"name": polars.String, "version": polars.String})
+            )
         else:
             helm_df = polars.read_json(StringIO(str(helm_out)))
             helm_df = helm_df.rename({"app_version": "version"})

src/edge_containers_cli/cmds/monitor.py

@@ -304,7 +304,7 @@ def _get_iocs(self) -> None:
         for i, ioc in enumerate(self.iocs):
             ioc = {key: value for key, value in ioc.items() if key not in exclude}
             self.iocs[i] = ioc
-        self.columns = [key for key in self.iocs[0].keys() if key not in exclude]
+        self.columns = [key for key in self.iocs_df.columns if key not in exclude]
 
     def _convert_df_to_list(self, iocs_df: polars.DataFrame | list) -> list[dict]:
         if isinstance(iocs_df, polars.DataFrame):