feat: added --missing flag to list and remove subcommands (#1056)

* feat: added `--missing` flag to list and remove subcommands

* chore: better fallback for missing kernels

Now use `shutil.which` to check missing kernels.  This
catches things like "python" or "bash"

* chore: added basic test to _limit_to_missing func

Author: wpk-nist-gov

jupyter_client/kernelspecapp.py

@@ -8,6 +8,7 @@
 import os.path
 import sys
 import typing as t
+from pathlib import Path
 
 from jupyter_core.application import JupyterApp, base_aliases, base_flags
 from traitlets import Bool, Dict, Instance, List, Unicode
@@ -29,12 +30,20 @@ class ListKernelSpecs(JupyterApp):
         help="output spec name and location as machine-readable json.",
         config=True,
     )
-
+    missing_kernels = Bool(
+        False,
+        help="List only specs with missing interpreters.",
+        config=True,
+    )
     flags = {
         "json": (
             {"ListKernelSpecs": {"json_output": True}},
             "output spec name and location as machine-readable json.",
         ),
+        "missing": (
+            {"ListKernelSpecs": {"missing_kernels": True}},
+            "output only missing kernels",
+        ),
         "debug": base_flags["debug"],
     }
 
@@ -45,6 +54,10 @@ def start(self) -> dict[str, t.Any] | None:  # type:ignore[override]
         """Start the application."""
         paths = self.kernel_spec_manager.find_kernel_specs()
         specs = self.kernel_spec_manager.get_all_specs()
+
+        if self.missing_kernels:
+            paths, specs = _limit_to_missing(paths, specs)
+
         if not self.json_output:
             if not specs:
                 print("No kernels available")
@@ -177,6 +190,11 @@ class RemoveKernelSpec(JupyterApp):
 
     force = Bool(False, config=True, help="""Force removal, don't prompt for confirmation.""")
     spec_names = List(Unicode())
+    missing_kernels = Bool(
+        False,
+        help="Remove missing specs.",
+        config=True,
+    )
 
     kernel_spec_manager = Instance(KernelSpecManager)
 
@@ -185,6 +203,10 @@ def _kernel_spec_manager_default(self) -> KernelSpecManager:
 
     flags = {
         "f": ({"RemoveKernelSpec": {"force": True}}, force.help),
+        "missing": (
+            {"RemoveKernelSpec": {"missing_kernels": True}},
+            "remove missing kernels",
+        ),
     }
     flags.update(JupyterApp.flags)
 
@@ -195,12 +217,22 @@ def parse_command_line(self, argv: list[str] | None) -> None:  # type:ignore[ove
         if self.extra_args:
             self.spec_names = sorted(set(self.extra_args))  # remove duplicates
         else:
-            self.exit("No kernelspec specified.")
+            self.spec_names = []
 
     def start(self) -> None:
         """Start the application."""
         self.kernel_spec_manager.ensure_native_kernel = False
         spec_paths = self.kernel_spec_manager.find_kernel_specs()
+
+        if self.missing_kernels:
+            _, spec = _limit_to_missing(
+                spec_paths,
+                self.kernel_spec_manager.get_all_specs(),
+            )
+
+            # append missing kernels
+            self.spec_names = sorted(set(self.spec_names + list(spec)))
+
         missing = set(self.spec_names).difference(set(spec_paths))
         if missing:
             self.exit("Couldn't find kernel spec(s): %s" % ", ".join(missing))
@@ -337,5 +369,22 @@ def start(self) -> None:
             return self.subapp.start()
 
 
+def _limit_to_missing(
+    paths: dict[str, str], specs: dict[str, t.Any]
+) -> tuple[dict[str, str], dict[str, t.Any]]:
+    from shutil import which
+
+    missing: dict[str, t.Any] = {}
+    for name, data in specs.items():
+        exe = data["spec"]["argv"][0]
+        # if exe exists or is on the path, keep it
+        if Path(exe).exists() or which(exe):
+            continue
+        missing[name] = data
+
+    paths_: dict[str, str] = {k: v for k, v in paths.items() if k in missing}
+    return paths_, missing
+
+
 if __name__ == "__main__":
     KernelSpecApp.launch_instance()

tests/test_kernelspecapp.py

@@ -3,6 +3,9 @@
 # Distributed under the terms of the Modified BSD License.
 import os
 import warnings
+from pathlib import Path
+
+import pytest
 
 from jupyter_client.kernelspecapp import (
     InstallKernelSpec,
@@ -38,6 +41,11 @@ def test_kernelspec_sub_apps(jp_kernel_dir):
     specs = app3.start()
     assert specs and "echo" not in specs
 
+    app4 = ListKernelSpecs(missing_kernels=True)
+    app4.kernel_spec_manager.kernel_dirs.append(kernel_dir)
+    specs = app4.start()
+    assert specs is None
+
 
 def test_kernelspec_app():
     app = KernelSpecApp()
@@ -49,3 +57,63 @@ def test_list_provisioners_app():
     app = ListProvisioners()
     app.initialize([])
     app.start()
+
+
+@pytest.fixture
+def dummy_kernelspecs():
+    import sys
+
+    p = Path.cwd().resolve()
+    # some missing kernelspecs
+    out = {
+        name: {
+            "resource_dir": str(p / name),
+            "spec": {
+                "argv": [
+                    str(p / name / "bin" / "python"),
+                    "-Xfrozen_modules=off",
+                    "-m",
+                    "ipykernel_launcher",
+                    "-f",
+                    "{connection_file}",
+                ],
+                "env": {},
+                "display_name": "Python [venv: dummy0]",
+                "language": "python",
+                "interrupt_mode": "signal",
+                "metadata": {"debugger": True},
+            },
+        }
+        for name in ("dummy0", "dummy1")
+    }
+
+    out["good"] = {
+        "resource_dir": str(p / "good"),
+        "spec": {
+            "argv": [
+                sys.executable,
+                "-Xfrozen_modules=off",
+                "-m",
+                "ipykernel_launcher",
+                "-f",
+                "{connection_file}",
+            ],
+            "env": {},
+            "display_name": "Python [venv: dummy0]",
+            "language": "python",
+            "interrupt_mode": "signal",
+            "metadata": {"debugger": True},
+        },
+    }
+    return out
+
+
+def test__limit_to_missing(dummy_kernelspecs) -> None:
+    from jupyter_client.kernelspecapp import _limit_to_missing
+
+    paths = {k: v["resource_dir"] for k, v in dummy_kernelspecs.items()}
+
+    paths, specs = _limit_to_missing(paths, dummy_kernelspecs)
+
+    assert specs == {k: v for k, v in dummy_kernelspecs.items() if k != "good"}
+    assert paths == {k: v["resource_dir"] for k, v in dummy_kernelspecs.items() if k != "good"}