deduplicate CLI subcommands, parser setup, and repo commands

lib/ramble/ramble/cmd/common/__init__.py

@@ -8,8 +8,18 @@
 
 import ramble.paths
 import ramble.util.colors as color
+from ramble.cmd.common.arguments import (
+    sanitize_arg_name,
+    setup_subcommands_from_prefix,
+)
 from ramble.util.logger import logger
 
+__all__ = [
+    "shell_init_instructions",
+    "sanitize_arg_name",
+    "setup_subcommands_from_prefix",
+]
+
 
 def shell_init_instructions(cmd, equivalent):
     """Print out instructions for users to initialize shell support.

lib/ramble/ramble/cmd/common/arguments.py

@@ -15,7 +15,13 @@
 
 from spack.util.pattern import Args
 
-__all__ = ["add_common_arguments", "allows_unknown_args", "validate_unknown_args"]
+__all__ = [
+    "add_common_arguments",
+    "allows_unknown_args",
+    "validate_unknown_args",
+    "sanitize_arg_name",
+    "setup_subcommands_from_prefix",
+]
 
 #: dictionary of argument-generating functions, keyed by name
 _arguments: Dict[str, Callable[[], Args]] = {}
@@ -238,3 +244,64 @@ def exclude_filter_group():
         help="Exclude experiments matching a logical expression of filter groups",
         required=False,
     )
+
+
+def sanitize_arg_name(base_name: str) -> str:
+    """Format argument/command names by converting hyphens to underscores."""
+    return base_name.replace("-", "_")
+
+
+def setup_subcommands_from_prefix(
+    subparser,
+    dest: str,
+    subcommands: list,
+    prefix: str,
+    globals_dict: dict,
+    subcommand_functions: dict,
+    inject_dry_run: bool = False,
+):
+    """Set up subparsers and map functions for a prefixed list of subcommands.
+
+    Args:
+        subparser: Parent ArgumentParser or subparser
+        dest (str): Destination attribute name for selected subcommand
+        subcommands (list): List of subcommand names or (name, *aliases) tuples
+        prefix (str): Prefix used in function names (e.g. 'workspace', 'data', 'deployment')
+        globals_dict (dict): The globals() dictionary of the calling module
+        subcommand_functions (dict): Dictionary mapping command/alias name -> function
+        inject_dry_run (bool): Whether to automatically add --dry-run to subparsers if not present
+    """
+    sp = subparser.add_subparsers(metavar="SUBCOMMAND", dest=dest)
+
+    for cmd_entry in subcommands:
+        if isinstance(cmd_entry, (list, tuple)):
+            name, aliases = cmd_entry[0], list(cmd_entry[1:])
+        else:
+            name = cmd_entry
+            aliases = []
+
+        # add commands to subcommands dict
+        function_name = sanitize_arg_name(f"{prefix}_{name}")
+        function = globals_dict[function_name]
+        for alias in [name] + aliases:
+            subcommand_functions[alias] = function
+
+        # make a subparser and run the command's setup function on it
+        setup_parser_cmd_name = sanitize_arg_name(f"{prefix}_{name}_setup_parser")
+        setup_parser_cmd = globals_dict[setup_parser_cmd_name]
+
+        subsubparser = sp.add_parser(
+            name,
+            aliases=aliases,
+            help=setup_parser_cmd.__doc__,
+            description=setup_parser_cmd.__doc__,
+        )
+        setup_parser_cmd(subsubparser)
+
+        if inject_dry_run and "--dry-run" not in subsubparser._option_string_actions:
+            subsubparser.add_argument(
+                "--dry-run",
+                dest="dry_run",
+                action="store_true",
+                help=f"perform a dry run of the {name} command",
+            )

lib/ramble/ramble/cmd/data.py

@@ -8,6 +8,8 @@
 
 from typing import Callable, Dict
 
+import ramble.cmd.common
+
 description = "manage data and databases"
 section = "data"
 level = "short"
@@ -60,39 +62,15 @@ def data_create_db(args):
 subcommand_functions: Dict[str, Callable] = {}
 
 
-def sanitize_arg_name(base_name):
-    """Allow function names to be remapped (eg `-` to `_`)"""
-    formatted_name = base_name.replace("-", "_")
-    return formatted_name
-
-
 def setup_parser(subparser):
-    sp = subparser.add_subparsers(metavar="SUBCOMMAND", dest="data_command")
-
-    for name in subcommands:
-        if isinstance(name, (list, tuple)):
-            name, aliases = name[0], name[1:]
-        else:
-            aliases = []
-
-        # add commands to subcommands dict
-        function_name = sanitize_arg_name(f"data_{name}")
-
-        function = globals()[function_name]
-        for alias in [name] + aliases:
-            subcommand_functions[alias] = function
-
-        # make a subparser and run the command's setup function on it
-        setup_parser_cmd_name = sanitize_arg_name(f"data_{name}_setup_parser")
-        setup_parser_cmd = globals()[setup_parser_cmd_name]
-
-        subsubparser = sp.add_parser(
-            name,
-            aliases=aliases,
-            help=setup_parser_cmd.__doc__,
-            description=setup_parser_cmd.__doc__,
-        )
-        setup_parser_cmd(subsubparser)
+    ramble.cmd.common.setup_subcommands_from_prefix(
+        subparser=subparser,
+        dest="data_command",
+        subcommands=subcommands,
+        prefix="data",
+        globals_dict=globals(),
+        subcommand_functions=subcommand_functions,
+    )
 
 
 def data(parser, args, unknown_args=None):

lib/ramble/ramble/cmd/deployment.py

@@ -12,6 +12,7 @@
 import llnl.util.filesystem as fs
 
 import ramble.cmd
+import ramble.cmd.common
 import ramble.config
 import ramble.fetch_strategy
 import ramble.filters
@@ -195,39 +196,15 @@ def deployment_run_pipeline(args, pipeline):
 subcommand_functions: Dict[str, Callable] = {}
 
 
-def sanitize_arg_name(base_name):
-    """Allow function names to be remapped (eg `-` to `_`)"""
-    formatted_name = base_name.replace("-", "_")
-    return formatted_name
-
-
 def setup_parser(subparser):
-    sp = subparser.add_subparsers(metavar="SUBCOMMAND", dest="deployment_command")
-
-    for name in subcommands:
-        if isinstance(name, (list, tuple)):
-            name, aliases = name[0], name[1:]
-        else:
-            aliases = []
-
-        # add commands to subcommands dict
-        function_name = sanitize_arg_name(f"deployment_{name}")
-
-        function = globals()[function_name]
-        for alias in [name] + aliases:
-            subcommand_functions[alias] = function
-
-        # make a subparser and run the command's setup function on it
-        setup_parser_cmd_name = sanitize_arg_name(f"deployment_{name}_setup_parser")
-        setup_parser_cmd = globals()[setup_parser_cmd_name]
-
-        subsubparser = sp.add_parser(
-            name,
-            aliases=aliases,
-            help=setup_parser_cmd.__doc__,
-            description=setup_parser_cmd.__doc__,
-        )
-        setup_parser_cmd(subsubparser)
+    ramble.cmd.common.setup_subcommands_from_prefix(
+        subparser=subparser,
+        dest="deployment_command",
+        subcommands=subcommands,
+        prefix="deployment",
+        globals_dict=globals(),
+        subcommand_functions=subcommand_functions,
+    )
 
 
 def deployment(parser, args):

lib/ramble/ramble/cmd/repo.py

@@ -121,6 +121,63 @@ def repo_create(args):
     )
 
 
+def _add_repo_to_config(obj_type, canon_path, original_path, namespace, scope):
+    """Add a canonical repository path for an object type to configuration
+    if not already present."""
+    type_def = ramble.repository.type_definitions[obj_type]
+    repos = ramble.config.get(type_def["config_section"], scope=scope) or []
+
+    is_present = any(
+        ramble.util.path.canonicalize_path(r_path) == canon_path or r_path == original_path
+        for r_path in repos
+    )
+
+    if is_present:
+        logger.warn(
+            f"{obj_type.name} repository is already registered with Ramble: {original_path}"
+        )
+    else:
+        repos.insert(0, canon_path)
+        ramble.config.set(type_def["config_section"], repos, scope)
+        logger.msg(f"Added {obj_type.name} repo with namespace '{namespace}'.")
+
+
+def _remove_repo_from_config(obj_type, namespace_or_path, scope):
+    """Remove a repository from a specific object type and scope by path or namespace."""
+    type_def = ramble.repository.type_definitions[obj_type]
+    repos = ramble.config.get(type_def["config_section"], scope=scope)
+    if not repos:
+        return False
+
+    canon_path = ramble.util.path.canonicalize_path(namespace_or_path)
+    normalized_path_to_remove = os.path.normcase(os.path.normpath(canon_path))
+
+    for repo_path in repos:
+        repo_canon_path = ramble.util.path.canonicalize_path(repo_path)
+        normalized_repo_path = os.path.normcase(os.path.normpath(repo_canon_path))
+        if normalized_path_to_remove == normalized_repo_path:
+            repos.remove(repo_path)
+            ramble.config.set(type_def["config_section"], repos, scope)
+            logger.msg(f"Removed {obj_type.name} repository {repo_path} from scope '{scope}'.")
+            return True
+
+    for path in list(repos):
+        try:
+            repo = ramble.repository.Repo(path, obj_type)
+            if repo.namespace == namespace_or_path:
+                repos.remove(path)
+                ramble.config.set(type_def["config_section"], repos, scope)
+                logger.msg(
+                    f"Removed {obj_type.name} repository {repo.root} "
+                    f"with namespace '{repo.namespace}' from scope '{scope}'."
+                )
+                return True
+        except ramble.repository.RepoError:
+            continue
+
+    return False
+
+
 def repo_add(args):
     """Add a repository to Ramble's configuration."""
     path = args.path
@@ -199,27 +256,9 @@ def repo_add(args):
 
         # Now add the canonical path for all object types
         for obj_type in ramble.repository.ObjectTypes:
-            type_def = ramble.repository.type_definitions[obj_type]
-            repos = ramble.config.get(type_def["config_section"], scope=args.scope) or []
-
-            # Check if canonical path is already present in the list of repos for this type
-            is_present = False
-            for r_path in repos:
-                if ramble.util.path.canonicalize_path(r_path) == canon_path:
-                    is_present = True
-                    break
-
-            if is_present:
-                logger.warn(
-                    f"{obj_type.name} repository is already registered with Ramble: {path}"
-                )
-            else:
-                repos.insert(0, canon_path)
-                ramble.config.set(type_def["config_section"], repos, args.scope)
-                logger.msg(f"Added {obj_type.name} repo with namespace '{repo_namespace}'.")
+            _add_repo_to_config(obj_type, canon_path, path, repo_namespace, args.scope)
     else:  # This is the original logic for a specific type
         obj_type = ramble.repository.ObjectTypes[args.type]
-        type_def = ramble.repository.type_definitions[obj_type]
         allow_partial = False  # For specific type, we don't allow partial
 
         # Make sure it's actually a ramble repository by constructing it.
@@ -241,14 +280,7 @@ def repo_add(args):
                 f"The given path {path} is not a valid repo for type {obj_type.name}"
             )
 
-        repos = ramble.config.get(type_def["config_section"], scope=args.scope) or []
-
-        if repo.root in repos or path in repos:
-            logger.warn(f"{obj_type.name} repository is already registered with Ramble: {path}")
-        else:
-            repos.insert(0, canon_path)
-            ramble.config.set(type_def["config_section"], repos, args.scope)
-            logger.msg(f"Added {obj_type.name} repo with namespace '{repo.namespace}'.")
+        _add_repo_to_config(obj_type, canon_path, path, repo.namespace, args.scope)
 
 
 def repo_remove(args):
@@ -267,47 +299,8 @@ def repo_remove(args):
     repo_removed = False
     for scope in scopes_to_check:
         for obj_type in obj_types:
-            type_def = ramble.repository.type_definitions[obj_type]
-            repos = ramble.config.get(type_def["config_section"], scope=scope)
-            if not repos:
-                continue
-
-            namespace_or_path = args.namespace_or_path
-
-            canon_path = ramble.util.path.canonicalize_path(namespace_or_path)
-            normalized_path_to_remove = os.path.normcase(os.path.normpath(canon_path))
-
-            path_found_and_removed = False
-            for repo_path in repos:
-                repo_canon_path = ramble.util.path.canonicalize_path(repo_path)
-                normalized_repo_path = os.path.normcase(os.path.normpath(repo_canon_path))
-                if normalized_path_to_remove == normalized_repo_path:
-                    repos.remove(repo_path)
-                    ramble.config.set(type_def["config_section"], repos, scope)
-                    logger.msg(
-                        f"Removed {obj_type.name} repository {repo_path} from scope '{scope}'."
-                    )
-                    repo_removed = True
-                    path_found_and_removed = True
-                    break  # move to next obj_type
-
-            if path_found_and_removed:
-                continue
-
-            for path in list(repos):
-                try:
-                    repo = ramble.repository.Repo(path, obj_type)
-                    if repo.namespace == namespace_or_path:
-                        repos.remove(path)
-                        ramble.config.set(type_def["config_section"], repos, scope)
-                        logger.msg(
-                            f"Removed {obj_type.name} repository {repo.root} "
-                            f"with namespace '{repo.namespace}' from scope '{scope}'."
-                        )
-                        repo_removed = True
-                        break
-                except ramble.repository.RepoError:
-                    continue
+            if _remove_repo_from_config(obj_type, args.namespace_or_path, scope):
+                repo_removed = True
 
         if repo_removed and not args.scope:
             break