Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Tools for managing kernel specs"""
3# Copyright (c) Jupyter Development Team.
4# Distributed under the terms of the Modified BSD License.
5from __future__ import annotations
7import json
8import os
9import re
10import shutil
11import typing as t
12import warnings
14from jupyter_core.paths import SYSTEM_JUPYTER_PATH, jupyter_data_dir, jupyter_path
15from traitlets import Bool, CaselessStrEnum, Dict, HasTraits, List, Set, Type, Unicode, observe
16from traitlets.config import LoggingConfigurable
18from .provisioning import KernelProvisionerFactory as KPF # noqa
20pjoin = os.path.join
22NATIVE_KERNEL_NAME = "python3"
25class KernelSpec(HasTraits):
26 """A kernel spec model object."""
28 argv: List[str] = List()
29 name = Unicode()
30 mimetype = Unicode()
31 display_name = Unicode()
32 language = Unicode()
33 kernel_protocol_version = Unicode()
34 env = Dict()
35 resource_dir = Unicode()
36 interrupt_mode = CaselessStrEnum(["message", "signal"], default_value="signal")
37 metadata = Dict()
39 @classmethod
40 def from_resource_dir(cls: type[KernelSpec], resource_dir: str) -> KernelSpec:
41 """Create a KernelSpec object by reading kernel.json
43 Pass the path to the *directory* containing kernel.json.
44 """
45 kernel_file = pjoin(resource_dir, "kernel.json")
46 with open(kernel_file, encoding="utf-8") as f:
47 kernel_dict = json.load(f)
48 return cls(resource_dir=resource_dir, **kernel_dict)
50 def to_dict(self) -> dict[str, t.Any]:
51 """Convert the kernel spec to a dict."""
52 d = {
53 "argv": self.argv,
54 "env": self.env,
55 "display_name": self.display_name,
56 "language": self.language,
57 "interrupt_mode": self.interrupt_mode,
58 "metadata": self.metadata,
59 "kernel_protocol_version": self.kernel_protocol_version,
60 }
62 return d
64 def to_json(self) -> str:
65 """Serialise this kernelspec to a JSON object.
67 Returns a string.
68 """
69 return json.dumps(self.to_dict())
72_kernel_name_pat = re.compile(r"^[a-z0-9._\-]+$", re.IGNORECASE)
75def _is_valid_kernel_name(name: str) -> t.Any:
76 """Check that a kernel name is valid."""
77 # quote is not unicode-safe on Python 2
78 return _kernel_name_pat.match(name)
81_kernel_name_description = (
82 "Kernel names can only contain ASCII letters and numbers and these separators:"
83 " - . _ (hyphen, period, and underscore)."
84)
87def _is_kernel_dir(path: str) -> bool:
88 """Is ``path`` a kernel directory?"""
89 return os.path.isdir(path) and os.path.isfile(pjoin(path, "kernel.json"))
92def _list_kernels_in(dir: str | None) -> dict[str, str]:
93 """Return a mapping of kernel names to resource directories from dir.
95 If dir is None or does not exist, returns an empty dict.
96 """
97 if dir is None or not os.path.isdir(dir):
98 return {}
99 kernels = {}
100 for f in os.listdir(dir):
101 path = pjoin(dir, f)
102 if not _is_kernel_dir(path):
103 continue
104 key = f.lower()
105 if not _is_valid_kernel_name(key):
106 warnings.warn(
107 f"Invalid kernelspec directory name ({_kernel_name_description}): {path}",
108 stacklevel=3,
109 )
110 kernels[key] = path
111 return kernels
114class NoSuchKernel(KeyError): # noqa
115 """An error raised when there is no kernel of a give name."""
117 def __init__(self, name: str) -> None:
118 """Initialize the error."""
119 self.name = name
121 def __str__(self) -> str:
122 return f"No such kernel named {self.name}"
125class KernelSpecManager(LoggingConfigurable):
126 """A manager for kernel specs."""
128 kernel_spec_class = Type(
129 KernelSpec,
130 config=True,
131 help="""The kernel spec class. This is configurable to allow
132 subclassing of the KernelSpecManager for customized behavior.
133 """,
134 )
136 ensure_native_kernel = Bool(
137 True,
138 config=True,
139 help="""If there is no Python kernelspec registered and the IPython
140 kernel is available, ensure it is added to the spec list.
141 """,
142 )
144 data_dir = Unicode()
146 def _data_dir_default(self) -> str:
147 return jupyter_data_dir()
149 user_kernel_dir = Unicode()
151 def _user_kernel_dir_default(self) -> str:
152 return pjoin(self.data_dir, "kernels")
154 whitelist = Set(
155 config=True,
156 help="""Deprecated, use `KernelSpecManager.allowed_kernelspecs`
157 """,
158 )
159 allowed_kernelspecs = Set(
160 config=True,
161 help="""List of allowed kernel names.
163 By default, all installed kernels are allowed.
164 """,
165 )
166 kernel_dirs: List[str] = List(
167 help="""
168 List of kernel directories to search.
170 Directories earlier in the list take priority: when two directories
171 contain a kernelspec with the same name, the one from the earlier
172 directory wins. This matches the precedence of ``$JUPYTER_PATH``,
173 where the first entry has the highest priority (see
174 ``jupyter_path("kernels")``).
175 """
176 )
178 _deprecated_aliases = {
179 "whitelist": ("allowed_kernelspecs", "7.0"),
180 }
182 # Method copied from
183 # https://github.com/jupyterhub/jupyterhub/blob/d1a85e53dccfc7b1dd81b0c1985d158cc6b61820/jupyterhub/auth.py#L143-L161
184 @observe(*list(_deprecated_aliases))
185 def _deprecated_trait(self, change: t.Any) -> None:
186 """observer for deprecated traits"""
187 old_attr = change.name
188 new_attr, version = self._deprecated_aliases[old_attr]
189 new_value = getattr(self, new_attr)
190 if new_value != change.new:
191 # only warn if different
192 # protects backward-compatible config from warnings
193 # if they set the same value under both names
194 self.log.warning(
195 f"{self.__class__.__name__}.{old_attr} is deprecated in jupyter_client "
196 f"{version}, use {self.__class__.__name__}.{new_attr} instead"
197 )
198 setattr(self, new_attr, change.new)
200 def _kernel_dirs_default(self) -> list[str]:
201 dirs = jupyter_path("kernels")
202 # At some point, we should stop adding .ipython/kernels to the path,
203 # but the cost to keeping it is very small.
204 try:
205 # this should always be valid on IPython 3+
206 from IPython.paths import get_ipython_dir
208 dirs.append(os.path.join(get_ipython_dir(), "kernels"))
209 except ModuleNotFoundError:
210 pass
211 return dirs
213 def find_kernel_specs(self) -> dict[str, str]:
214 """Returns a dict mapping kernel names to resource directories."""
215 d = {}
216 for kernel_dir in self.kernel_dirs:
217 kernels = _list_kernels_in(kernel_dir)
218 for kname, spec in kernels.items():
219 if kname not in d:
220 self.log.debug("Found kernel %s in %s", kname, kernel_dir)
221 d[kname] = spec
223 if self.ensure_native_kernel and NATIVE_KERNEL_NAME not in d:
224 try:
225 from ipykernel.kernelspec import RESOURCES
227 self.log.debug(
228 "Native kernel (%s) available from %s",
229 NATIVE_KERNEL_NAME,
230 RESOURCES,
231 )
232 d[NATIVE_KERNEL_NAME] = RESOURCES
233 except ImportError:
234 self.log.warning("Native kernel (%s) is not available", NATIVE_KERNEL_NAME)
236 if self.allowed_kernelspecs:
237 # filter if there's an allow list
238 d = {name: spec for name, spec in d.items() if name in self.allowed_kernelspecs}
239 return d
240 # TODO: Caching?
242 def _get_kernel_spec_by_name(self, kernel_name: str, resource_dir: str) -> KernelSpec:
243 """Returns a :class:`KernelSpec` instance for a given kernel_name
244 and resource_dir.
245 """
246 kspec = None
247 if kernel_name == NATIVE_KERNEL_NAME:
248 try:
249 from ipykernel.kernelspec import RESOURCES, get_kernel_dict
250 except ImportError:
251 # It should be impossible to reach this, but let's play it safe
252 pass
253 else:
254 if resource_dir == RESOURCES:
255 kdict = get_kernel_dict()
256 kspec = self.kernel_spec_class(resource_dir=resource_dir, **kdict)
257 if not kspec:
258 kspec = self.kernel_spec_class.from_resource_dir(resource_dir)
260 if not KPF.instance(parent=self.parent).is_provisioner_available(kspec):
261 raise NoSuchKernel(kernel_name)
263 return kspec
265 def _find_spec_directory(self, kernel_name: str) -> str | None:
266 """Find the resource directory of a named kernel spec"""
267 for kernel_dir in [kd for kd in self.kernel_dirs if os.path.isdir(kd)]:
268 files = os.listdir(kernel_dir)
269 for f in files:
270 path = pjoin(kernel_dir, f)
271 if f.lower() == kernel_name and _is_kernel_dir(path):
272 return path
274 if kernel_name == NATIVE_KERNEL_NAME:
275 try:
276 from ipykernel.kernelspec import RESOURCES
277 except ImportError:
278 pass
279 else:
280 return RESOURCES
281 return None
283 def get_kernel_spec(self, kernel_name: str) -> KernelSpec:
284 """Returns a :class:`KernelSpec` instance for the given kernel_name.
286 Raises :exc:`NoSuchKernel` if the given kernel name is not found.
287 """
288 if not _is_valid_kernel_name(kernel_name):
289 self.log.warning(
290 f"Kernelspec name {kernel_name} is invalid: {_kernel_name_description}"
291 )
293 resource_dir = self._find_spec_directory(kernel_name.lower())
294 if resource_dir is None:
295 raise NoSuchKernel(kernel_name)
297 return self._get_kernel_spec_by_name(kernel_name, resource_dir)
299 def get_all_specs(self) -> dict[str, t.Any]:
300 """Returns a dict mapping kernel names to kernelspecs.
302 Returns a dict of the form::
304 {
305 'kernel_name': {
306 'resource_dir': '/path/to/kernel_name',
307 'spec': {"the spec itself": ...}
308 },
309 ...
310 }
311 """
312 d = self.find_kernel_specs()
313 res = {}
314 for kname, resource_dir in d.items():
315 try:
316 if self.__class__ is KernelSpecManager:
317 spec = self._get_kernel_spec_by_name(kname, resource_dir)
318 else:
319 # avoid calling private methods in subclasses,
320 # which may have overridden find_kernel_specs
321 # and get_kernel_spec, but not the newer get_all_specs
322 spec = self.get_kernel_spec(kname)
324 res[kname] = {"resource_dir": resource_dir, "spec": spec.to_dict()}
325 except NoSuchKernel:
326 pass # The appropriate warning has already been logged
327 except Exception:
328 self.log.warning("Error loading kernelspec %r", kname, exc_info=True)
329 return res
331 def remove_kernel_spec(self, name: str) -> str:
332 """Remove a kernel spec directory by name.
334 Returns the path that was deleted.
335 """
336 save_native = self.ensure_native_kernel
337 try:
338 self.ensure_native_kernel = False
339 specs = self.find_kernel_specs()
340 finally:
341 self.ensure_native_kernel = save_native
342 spec_dir = specs[name]
343 self.log.debug("Removing %s", spec_dir)
344 if os.path.islink(spec_dir):
345 os.remove(spec_dir)
346 else:
347 shutil.rmtree(spec_dir)
348 return spec_dir
350 def _get_destination_dir(
351 self, kernel_name: str, user: bool = False, prefix: str | None = None
352 ) -> str:
353 if user:
354 return os.path.join(self.user_kernel_dir, kernel_name)
355 elif prefix:
356 return os.path.join(os.path.abspath(prefix), "share", "jupyter", "kernels", kernel_name)
357 else:
358 return os.path.join(SYSTEM_JUPYTER_PATH[0], "kernels", kernel_name)
360 def install_kernel_spec(
361 self,
362 source_dir: str,
363 kernel_name: str | None = None,
364 user: bool = False,
365 replace: bool | None = None,
366 prefix: str | None = None,
367 ) -> str:
368 """Install a kernel spec by copying its directory.
370 If ``kernel_name`` is not given, the basename of ``source_dir`` will
371 be used.
373 If ``user`` is False, it will attempt to install into the systemwide
374 kernel registry. If the process does not have appropriate permissions,
375 an :exc:`OSError` will be raised.
377 If ``prefix`` is given, the kernelspec will be installed to
378 PREFIX/share/jupyter/kernels/KERNEL_NAME. This can be sys.prefix
379 for installation inside virtual or conda envs.
380 """
381 source_dir = source_dir.rstrip("/\\")
382 if not kernel_name:
383 kernel_name = os.path.basename(source_dir)
384 kernel_name = kernel_name.lower()
385 if not _is_valid_kernel_name(kernel_name):
386 msg = f"Invalid kernel name {kernel_name!r}. {_kernel_name_description}"
387 raise ValueError(msg)
389 if user and prefix:
390 msg = "Can't specify both user and prefix. Please choose one or the other."
391 raise ValueError(msg)
393 if replace is not None:
394 warnings.warn(
395 "replace is ignored. Installing a kernelspec always replaces an existing "
396 "installation",
397 DeprecationWarning,
398 stacklevel=2,
399 )
401 destination = self._get_destination_dir(kernel_name, user=user, prefix=prefix)
402 self.log.debug("Installing kernelspec in %s", destination)
404 kernel_dir = os.path.dirname(destination)
405 if kernel_dir not in self.kernel_dirs:
406 self.log.warning(
407 "Installing to %s, which is not in %s. The kernelspec may not be found.",
408 kernel_dir,
409 self.kernel_dirs,
410 )
412 if os.path.isdir(destination):
413 self.log.info("Removing existing kernelspec in %s", destination)
414 shutil.rmtree(destination)
416 shutil.copytree(source_dir, destination)
417 self.log.info("Installed kernelspec %s in %s", kernel_name, destination)
418 return destination
420 def install_native_kernel_spec(self, user: bool = False) -> None:
421 """DEPRECATED: Use ipykernel.kernelspec.install"""
422 warnings.warn(
423 "install_native_kernel_spec is deprecated. Use ipykernel.kernelspec import install.",
424 stacklevel=2,
425 )
426 from ipykernel.kernelspec import install
428 install(self, user=user)
431def find_kernel_specs() -> dict[str, str]:
432 """Returns a dict mapping kernel names to resource directories."""
433 return KernelSpecManager().find_kernel_specs()
436def get_kernel_spec(kernel_name: str) -> KernelSpec:
437 """Returns a :class:`KernelSpec` instance for the given kernel_name.
439 Raises KeyError if the given kernel name is not found.
440 """
441 return KernelSpecManager().get_kernel_spec(kernel_name)
444def install_kernel_spec(
445 source_dir: str,
446 kernel_name: str | None = None,
447 user: bool = False,
448 replace: bool | None = False,
449 prefix: str | None = None,
450) -> str:
451 """Install a kernel spec in a given directory."""
452 return KernelSpecManager().install_kernel_spec(source_dir, kernel_name, user, replace, prefix)
455install_kernel_spec.__doc__ = KernelSpecManager.install_kernel_spec.__doc__
458def install_native_kernel_spec(user: bool = False) -> None:
459 """Install the native kernel spec."""
460 KernelSpecManager().install_native_kernel_spec(user=user)
463install_native_kernel_spec.__doc__ = KernelSpecManager.install_native_kernel_spec.__doc__