fix: move get_current_namespace to utils dir

Author: kryanbeane

src/codeflare_sdk/common/kueue/kueue.py

@@ -18,6 +18,8 @@
 from kubernetes import client
 from kubernetes.client.exceptions import ApiException
 
+from ...common.utils import get_current_namespace
+
 
 def get_default_kueue_name(namespace: str) -> Optional[str]:
     """
@@ -81,7 +83,6 @@ def list_local_queues(
         List[dict]:
             A list of dictionaries containing the name of the local queue and the available flavors
     """
-    from ...ray.cluster.cluster import get_current_namespace
 
     if namespace is None:  # pragma: no cover
         namespace = get_current_namespace()

src/codeflare_sdk/common/utils/__init__.py

@@ -0,0 +1,7 @@
+"""
+Common utilities for the CodeFlare SDK.
+"""
+
+from .k8s_utils import get_current_namespace
+
+__all__ = ["get_current_namespace"]

src/codeflare_sdk/common/utils/k8s_utils.py

@@ -0,0 +1,37 @@
+"""
+Kubernetes utility functions for the CodeFlare SDK.
+"""
+
+import os
+from kubernetes import config
+from ..kubernetes_cluster import config_check, _kube_api_error_handling
+
+
+def get_current_namespace():
+    """
+    Retrieves the current Kubernetes namespace.
+
+    This function attempts to detect the current namespace by:
+    1. First checking if running inside a pod (reading from service account namespace file)
+    2. Falling back to reading from the current kubeconfig context
+
+    Returns:
+        str:
+            The current namespace or None if not found.
+    """
+    if os.path.isfile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"):
+        try:
+            file = open("/var/run/secrets/kubernetes.io/serviceaccount/namespace", "r")
+            active_context = file.readline().strip("\n")
+            return active_context
+        except Exception as e:
+            print("Unable to find current namespace")
+    print("trying to gather from current context")
+    try:
+        _, active_context = config.list_kube_config_contexts(config_check())
+    except Exception as e:
+        return _kube_api_error_handling(e)
+    try:
+        return active_context["context"]["namespace"]
+    except KeyError:
+        return None

src/codeflare_sdk/common/widgets/test_widgets.py

@@ -106,7 +106,7 @@ def test_view_clusters(mocker, capsys):
     # Prepare to run view_clusters when notebook environment is detected
     mocker.patch("codeflare_sdk.common.widgets.widgets.is_notebook", return_value=True)
     mock_get_current_namespace = mocker.patch(
-        "codeflare_sdk.ray.cluster.cluster.get_current_namespace",
+        "codeflare_sdk.common.utils.get_current_namespace",
         return_value="default",
     )
     namespace = mock_get_current_namespace.return_value
@@ -250,7 +250,7 @@ def test_ray_cluster_manager_widgets_init(mocker, capsys):
         return_value=test_ray_clusters_df,
     )
     mocker.patch(
-        "codeflare_sdk.ray.cluster.cluster.get_current_namespace",
+        "codeflare_sdk.common.utils.get_current_namespace",
         return_value=namespace,
     )
     mock_delete_cluster = mocker.patch(

src/codeflare_sdk/common/widgets/widgets.py

@@ -26,6 +26,8 @@
 import ipywidgets as widgets
 from IPython.display import display, HTML, Javascript
 import pandas as pd
+
+from ...common.utils import get_current_namespace
 from ...ray.cluster.config import ClusterConfiguration
 from ...ray.cluster.status import RayClusterStatus
 from ..kubernetes_cluster import _kube_api_error_handling
@@ -43,8 +45,6 @@ class RayClusterManagerWidgets:
     """
 
     def __init__(self, ray_clusters_df: pd.DataFrame, namespace: str = None):
-        from ...ray.cluster.cluster import get_current_namespace
-
         # Data
         self.ray_clusters_df = ray_clusters_df
         self.namespace = get_current_namespace() if not namespace else namespace
@@ -353,7 +353,7 @@ def view_clusters(namespace: str = None):
         )
         return  # Exit function if not in Jupyter Notebook
 
-    from ...ray.cluster.cluster import get_current_namespace
+    from ...common.utils import get_current_namespace
 
     if not namespace:
         namespace = get_current_namespace()

src/codeflare_sdk/ray/cluster/cluster.py

@@ -27,6 +27,8 @@
 import uuid
 import warnings
 
+from ...common.utils import get_current_namespace
+
 from ...common.kubernetes_cluster.auth import (
     config_check,
     get_api_client,
@@ -638,32 +640,6 @@ def list_all_queued(
     return resources
 
 
-def get_current_namespace():  # pragma: no cover
-    """
-    Retrieves the current Kubernetes namespace.
-
-    Returns:
-        str:
-            The current namespace or None if not found.
-    """
-    if os.path.isfile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"):
-        try:
-            file = open("/var/run/secrets/kubernetes.io/serviceaccount/namespace", "r")
-            active_context = file.readline().strip("\n")
-            return active_context
-        except Exception as e:
-            print("Unable to find current namespace")
-    print("trying to gather from current context")
-    try:
-        _, active_context = config.list_kube_config_contexts(config_check())
-    except Exception as e:
-        return _kube_api_error_handling(e)
-    try:
-        return active_context["context"]["namespace"]
-    except KeyError:
-        return None
-
-
 def get_cluster(
     cluster_name: str,
     namespace: str = "default",