Consolidating argument processing into standardized function

ktbyers · Oct 1, 2024 · e53ef6d · e53ef6d
1 parent 23fa073
commit e53ef6d
Show file tree

Hide file tree

Showing 4 changed files with 50 additions and 75 deletions.
diff --git a/netmiko/cli_tools/argument_handling.py b/netmiko/cli_tools/argument_handling.py
@@ -1,4 +1,6 @@
 import argparse
+from getpass import getpass
+from netmiko.utilities import load_devices, display_inventory
 
 
 def common_args(parser):
@@ -96,3 +98,22 @@ def parse_arguments(args, command):
             parser.error("Grep pattern not specified.")
 
     return cli_args
+
+
+def extract_cli_vars(cli_args, command, __version__):
+
+    return_vars = {}
+    return_vars["cli_username"] = cli_args.username if cli_args.username else None
+    return_vars["cli_password"] = getpass() if cli_args.password else None
+    return_vars["cli_secret"] = getpass("Enable secret: ") if cli_args.secret else None
+    version = cli_args.version
+    if version:
+        print(f"{command} v{__version__}")
+        return 0
+    list_devices = cli_args.list_devices
+    if list_devices:
+        my_devices = load_devices()
+        display_inventory(my_devices)
+        return 0
+
+    return return_vars
diff --git a/netmiko/cli_tools/netmiko_cfg.py b/netmiko/cli_tools/netmiko_cfg.py
@@ -2,14 +2,12 @@
 """Return output from single show cmd using Netmiko."""
 import sys
 from datetime import datetime
-from getpass import getpass
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
-from netmiko.utilities import load_devices, display_inventory
 from netmiko.cli_tools import ERROR_PATTERN, MAX_WORKERS, __version__
 from netmiko.cli_tools.helpers import obtain_devices, update_device_params, ssh_conn
 from netmiko.cli_tools.outputters import output_dispatcher, output_failed_devices
-from netmiko.cli_tools.argument_handling import parse_arguments
+from netmiko.cli_tools.argument_handling import parse_arguments, extract_cli_vars
 
 
 COMMAND = "netmiko-cfg"
@@ -21,24 +19,14 @@ def main_ep():
 
 def main(args):
     start_time = datetime.now()
+
+    # CLI ARGS #####
     cli_args = parse_arguments(args, COMMAND)
+    cli_vars = extract_cli_vars(cli_args, command=COMMAND, __version__=__version__)
+    device_or_group = cli_args.devices.strip()
+    hide_failed = cli_args.hide_failed
 
-    cli_username = cli_args.username if cli_args.username else None
-    cli_password = getpass() if cli_args.password else None
-    cli_secret = getpass("Enable secret: ") if cli_args.secret else None
-
-    version = cli_args.version
-    if version:
-        print(f"{COMMAND} v{__version__}")
-        return 0
-    list_devices = cli_args.list_devices
-    if list_devices:
-        my_devices = load_devices()
-        display_inventory(my_devices)
-        return 0
-
-    output_json = cli_args.json
-    output_raw = cli_args.raw
+    # CFG COMMAND HANDLER #####
     cfg_command = cli_args.cmd
     if cfg_command:
         if r"\n" in cfg_command:
@@ -50,8 +38,6 @@ def main(args):
         cfg_command = command_data.splitlines()
     else:
         raise ValueError("No configuration commands provided.")
-    device_or_group = cli_args.devices.strip()
-    hide_failed = cli_args.hide_failed
 
     # DEVICE LOADING #####
     devices = obtain_devices(device_or_group)
@@ -65,9 +51,9 @@ def main(args):
     for device_name, device_params in devices.items():
         update_device_params(
             device_params,
-            username=cli_username,
-            password=cli_password,
-            secret=cli_secret,
+            username=cli_vars["cli_username"],
+            password=cli_vars["cli_password"],
+            secret=cli_vars["cli_secret"],
         )
         device_tasks.append(
             {
@@ -97,11 +83,11 @@ def main(args):
 
     # OUTPUT PROCESSING #####
     out_format = "text"
-    if output_json and output_raw:
+    if cli_args.json and cli_args.raw:
         out_format = "json_raw"
-    elif output_json:
+    elif cli_args.json:
         out_format = "json"
-    elif output_raw:
+    elif cli_args.raw:
         out_format = "raw"
     # elif output_yaml:
     #    out_format = "yaml"

diff --git a/netmiko/cli_tools/netmiko_grep.py b/netmiko/cli_tools/netmiko_grep.py
@@ -5,15 +5,13 @@
 import subprocess
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime
-from getpass import getpass
 
-from netmiko.utilities import load_devices, display_inventory
 from netmiko.utilities import write_tmp_file, ensure_dir_exists
 from netmiko.utilities import find_netmiko_dir
 from netmiko.utilities import SHOW_RUN_MAPPER
 from netmiko.cli_tools import ERROR_PATTERN, GREP, MAX_WORKERS, __version__
 from netmiko.cli_tools.helpers import obtain_devices, update_device_params, ssh_conn
-from netmiko.cli_tools.argument_handling import parse_arguments
+from netmiko.cli_tools.argument_handling import parse_arguments, extract_cli_vars
 
 
 COMMAND = "netmiko-grep"
@@ -45,29 +43,17 @@ def main_ep():
 
 def main(args):
     start_time = datetime.now()
-    cli_args = parse_arguments(args, COMMAND)
-
-    cli_username = cli_args.username if cli_args.username else None
-    cli_password = getpass() if cli_args.password else None
-    cli_secret = getpass("Enable secret: ") if cli_args.secret else None
-
-    version = cli_args.version
-    if version:
-        print(f"{COMMAND} v{__version__}")
-        return 0
-    list_devices = cli_args.list_devices
-    if list_devices:
-        my_devices = load_devices()
-        display_inventory(my_devices)
-        return 0
 
+    # CLI ARGS #####
+    cli_args = parse_arguments(args, COMMAND)
+    cli_vars = extract_cli_vars(cli_args, command=COMMAND, __version__=__version__)
     cli_command = cli_args.cmd
     cmd_arg = False
     if cli_command:
         cmd_arg = True
     device_or_group = cli_args.devices.strip()
-    pattern = cli_args.pattern
     hide_failed = cli_args.hide_failed
+    pattern = cli_args.pattern
 
     # DEVICE LOADING #####
     devices = obtain_devices(device_or_group)
@@ -82,9 +68,9 @@ def main(args):
     for device_name, device_params in devices.items():
         update_device_params(
             device_params,
-            username=cli_username,
-            password=cli_password,
-            secret=cli_secret,
+            username=cli_vars["cli_username"],
+            password=cli_vars["cli_password"],
+            secret=cli_vars["cli_secret"],
         )
         if not cmd_arg:
             device_type = device_params["device_type"]

diff --git a/netmiko/cli_tools/netmiko_show.py b/netmiko/cli_tools/netmiko_show.py
@@ -4,15 +4,13 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
 from datetime import datetime
-from getpass import getpass
 from rich import print
 
-from netmiko.utilities import load_devices, display_inventory
 from netmiko.utilities import SHOW_RUN_MAPPER
 from netmiko.cli_tools import ERROR_PATTERN, MAX_WORKERS, __version__
 from netmiko.cli_tools.helpers import obtain_devices, update_device_params, ssh_conn
 from netmiko.cli_tools.outputters import output_dispatcher, output_failed_devices
-from netmiko.cli_tools.argument_handling import parse_arguments
+from netmiko.cli_tools.argument_handling import parse_arguments, extract_cli_vars
 
 
 COMMAND = "netmiko-show"
@@ -27,23 +25,7 @@ def main(args):
 
     # CLI ARGS #####
     cli_args = parse_arguments(args, COMMAND)
-
-    cli_username = cli_args.username if cli_args.username else None
-    cli_password = getpass() if cli_args.password else None
-    cli_secret = getpass("Enable secret: ") if cli_args.secret else None
-
-    version = cli_args.version
-    if version:
-        print(f"{COMMAND} v{__version__}")
-        return 0
-    list_devices = cli_args.list_devices
-    if list_devices:
-        my_devices = load_devices()
-        display_inventory(my_devices)
-        return 0
-
-    output_json = cli_args.json
-    output_raw = cli_args.raw
+    cli_vars = extract_cli_vars(cli_args, command=COMMAND, __version__=__version__)
     cli_command = cli_args.cmd
     cmd_arg = False
     if cli_command:
@@ -63,9 +45,9 @@ def main(args):
     for device_name, device_params in devices.items():
         update_device_params(
             device_params,
-            username=cli_username,
-            password=cli_password,
-            secret=cli_secret,
+            username=cli_vars["cli_username"],
+            password=cli_vars["cli_password"],
+            secret=cli_vars["cli_secret"],
         )
         if not cmd_arg:
             device_type = device_params["device_type"]
@@ -98,11 +80,11 @@ def main(args):
 
     # OUTPUT PROCESSING #####
     out_format = "text"
-    if output_json and output_raw:
+    if cli_args.json and cli_args.raw:
         out_format = "json_raw"
-    elif output_json:
+    elif cli_args.json:
         out_format = "json"
-    elif output_raw:
+    elif cli_args.raw:
         out_format = "raw"
     # elif output_yaml:
     #    out_format = "yaml"