diff --git a/jupyter_client/kernelspecapp.py b/jupyter_client/kernelspecapp.py index 186aa9cf..ceb730b0 100644 --- a/jupyter_client/kernelspecapp.py +++ b/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() diff --git a/tests/test_kernelspecapp.py b/tests/test_kernelspecapp.py index 119b6fb6..ae2c0a00 100644 --- a/tests/test_kernelspecapp.py +++ b/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"}