XpkConfig as a static singleton (#800)

* feat: XpkConfig as a static singleton

fix: import

* style: rename xpk_config_instance -> xpk_config

Author: scaliby

src/xpk/commands/config.py

@@ -14,16 +14,14 @@
 limitations under the License.
 """
 
-from ..core.config import XpkConfig
+from ..core.config import xpk_config
 from ..utils.console import xpk_print
 
-xpk_cfg = XpkConfig()
-
 
 def set_config(args):
-  xpk_cfg.set(args.set_config_args[0], args.set_config_args[1])
+  xpk_config.set(args.set_config_args[0], args.set_config_args[1])
 
 
 def get_config(args):
-  value = xpk_cfg.get(args.get_config_key[0])
+  value = xpk_config.get(args.get_config_key[0])
   xpk_print(value)

src/xpk/core/config.py

@@ -114,3 +114,6 @@ def get_all(
       return None
     val: dict[str, str] = config_yaml[CONFIGS_KEY]
     return val
+
+
+xpk_config = XpkConfig()

src/xpk/core/kjob.py

@@ -38,7 +38,7 @@
     KJOB_SHELL_IMAGE,
     KJOB_SHELL_INTERACTIVE_COMMAND,
     KJOB_SHELL_WORKING_DIRECTORY,
-    XpkConfig,
+    xpk_config,
 )
 from .network import get_cluster_subnetworks
 from .system_characteristics import AcceleratorType, SystemCharacteristics
@@ -234,8 +234,7 @@ def get_pod_template_interactive_command() -> str:
   Returns:
     str - PodTemplate's interactive command
   """
-  config = XpkConfig()
-  pod_command = config.get(KJOB_SHELL_INTERACTIVE_COMMAND)
+  pod_command = xpk_config.get(KJOB_SHELL_INTERACTIVE_COMMAND)
   if pod_command is None or len(pod_command) == 0:
     pod_command = PodTemplateDefaults.INTERACTIVE_COMMAND.value
 
@@ -287,11 +286,10 @@ def create_job_template_instance(
   Returns:
     exit_code > 0 if creating JobTemplate fails, 0 otherwise
   """
-  config = XpkConfig()
-  job_image = config.get(KJOB_BATCH_IMAGE)
+  job_image = xpk_config.get(KJOB_BATCH_IMAGE)
   if job_image is None or len(job_image) == 0:
     job_image = JobTemplateDefaults.IMAGE.value
-  working_directory = config.get(KJOB_BATCH_WORKING_DIRECTORY)
+  working_directory = xpk_config.get(KJOB_BATCH_WORKING_DIRECTORY)
   if working_directory is None or len(working_directory) == 0:
     working_directory = JobTemplateDefaults.WORKING_DIRECTORY.value
   resources = (
@@ -332,11 +330,10 @@ def create_pod_template_instance(service_account: str) -> int:
   Returns:
     exit_code > 0 if creating PodTemplate fails, 0 otherwise
   """
-  config = XpkConfig()
-  pod_image = config.get(KJOB_SHELL_IMAGE)
+  pod_image = xpk_config.get(KJOB_SHELL_IMAGE)
   if pod_image is None or len(pod_image) == 0:
     pod_image = PodTemplateDefaults.IMAGE.value
-  working_directory = config.get(KJOB_SHELL_WORKING_DIRECTORY)
+  working_directory = xpk_config.get(KJOB_SHELL_WORKING_DIRECTORY)
   if working_directory is None or len(working_directory) == 0:
     working_directory = PodTemplateDefaults.WORKING_DIRECTORY.value
 

src/xpk/parser/cluster.py

@@ -26,7 +26,7 @@
     cluster_describe,
     cluster_list,
 )
-from ..commands.config import xpk_cfg
+from ..core.config import xpk_config
 from ..core.system_characteristics import get_system_characteristics_keys_by_accelerator_type, AcceleratorType
 from ..core.config import CFG_BUCKET_KEY
 from ..core.vertex import DEFAULT_VERTEX_TENSORBOARD_NAME
@@ -131,7 +131,7 @@ def set_cluster_create_parser(cluster_create_parser: ArgumentParser):
   cluster_create_optional_arguments.add_argument(
       '--cluster-state-gcs-bucket',
       type=str,
-      default=xpk_cfg.get(CFG_BUCKET_KEY),
+      default=xpk_config.get(CFG_BUCKET_KEY),
       help='The name of the bucket to store cluster state.',
       required=False,
   )
@@ -390,7 +390,7 @@ def set_cluster_delete_parser(cluster_delete_parser: ArgumentParser):
   cluster_delete_optional_arguments.add_argument(
       '--cluster-state-gcs-bucket',
       type=str,
-      default=xpk_cfg.get(CFG_BUCKET_KEY),
+      default=xpk_config.get(CFG_BUCKET_KEY),
       help='The name of the bucket to store cluster state.',
       required=False,
   )