Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/jupyter_client/kernelspec.py: 30%
208 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-07-01 06:54 +0000
1"""Tools for managing kernel specs"""
2# Copyright (c) Jupyter Development Team.
3# Distributed under the terms of the Modified BSD License.
4import json
5import os
6import re
7import shutil
8import warnings
10from jupyter_core.paths import SYSTEM_JUPYTER_PATH, jupyter_data_dir, jupyter_path
11from traitlets import Bool, CaselessStrEnum, Dict, HasTraits, List, Set, Type, Unicode, observe
12from traitlets.config import LoggingConfigurable
14from .provisioning import KernelProvisionerFactory as KPF # noqa
16pjoin = os.path.join
18NATIVE_KERNEL_NAME = "python3"
21class KernelSpec(HasTraits):
22 """A kernel spec model object."""
24 argv = List()
25 name = Unicode()
26 mimetype = Unicode()
27 display_name = Unicode()
28 language = Unicode()
29 env = Dict()
30 resource_dir = Unicode()
31 interrupt_mode = CaselessStrEnum(["message", "signal"], default_value="signal")
32 metadata = Dict()
34 @classmethod
35 def from_resource_dir(cls, resource_dir):
36 """Create a KernelSpec object by reading kernel.json
38 Pass the path to the *directory* containing kernel.json.
39 """
40 kernel_file = pjoin(resource_dir, "kernel.json")
41 with open(kernel_file, encoding="utf-8") as f:
42 kernel_dict = json.load(f)
43 return cls(resource_dir=resource_dir, **kernel_dict)
45 def to_dict(self):
46 """Convert the kernel spec to a dict."""
47 d = {
48 "argv": self.argv,
49 "env": self.env,
50 "display_name": self.display_name,
51 "language": self.language,
52 "interrupt_mode": self.interrupt_mode,
53 "metadata": self.metadata,
54 }
56 return d
58 def to_json(self):
59 """Serialise this kernelspec to a JSON object.
61 Returns a string.
62 """
63 return json.dumps(self.to_dict())
66_kernel_name_pat = re.compile(r"^[a-z0-9._\-]+$", re.IGNORECASE)
69def _is_valid_kernel_name(name):
70 """Check that a kernel name is valid."""
71 # quote is not unicode-safe on Python 2
72 return _kernel_name_pat.match(name)
75_kernel_name_description = (
76 "Kernel names can only contain ASCII letters and numbers and these separators:"
77 " - . _ (hyphen, period, and underscore)."
78)
81def _is_kernel_dir(path):
82 """Is ``path`` a kernel directory?"""
83 return os.path.isdir(path) and os.path.isfile(pjoin(path, "kernel.json"))
86def _list_kernels_in(dir):
87 """Return a mapping of kernel names to resource directories from dir.
89 If dir is None or does not exist, returns an empty dict.
90 """
91 if dir is None or not os.path.isdir(dir):
92 return {}
93 kernels = {}
94 for f in os.listdir(dir):
95 path = pjoin(dir, f)
96 if not _is_kernel_dir(path):
97 continue
98 key = f.lower()
99 if not _is_valid_kernel_name(key):
100 warnings.warn(
101 f"Invalid kernelspec directory name ({_kernel_name_description}): {path}",
102 stacklevel=3,
103 )
104 kernels[key] = path
105 return kernels
108class NoSuchKernel(KeyError): # noqa
109 """An error raised when there is no kernel of a give name."""
111 def __init__(self, name):
112 """Initialize the error."""
113 self.name = name
115 def __str__(self):
116 return f"No such kernel named {self.name}"
119class KernelSpecManager(LoggingConfigurable):
120 """A manager for kernel specs."""
122 kernel_spec_class = Type(
123 KernelSpec,
124 config=True,
125 help="""The kernel spec class. This is configurable to allow
126 subclassing of the KernelSpecManager for customized behavior.
127 """,
128 )
130 ensure_native_kernel = Bool(
131 True,
132 config=True,
133 help="""If there is no Python kernelspec registered and the IPython
134 kernel is available, ensure it is added to the spec list.
135 """,
136 )
138 data_dir = Unicode()
140 def _data_dir_default(self):
141 return jupyter_data_dir()
143 user_kernel_dir = Unicode()
145 def _user_kernel_dir_default(self):
146 return pjoin(self.data_dir, "kernels")
148 whitelist = Set(
149 config=True,
150 help="""Deprecated, use `KernelSpecManager.allowed_kernelspecs`
151 """,
152 )
153 allowed_kernelspecs = Set(
154 config=True,
155 help="""List of allowed kernel names.
157 By default, all installed kernels are allowed.
158 """,
159 )
160 kernel_dirs = List(
161 help="List of kernel directories to search. Later ones take priority over earlier."
162 )
164 _deprecated_aliases = {
165 "whitelist": ("allowed_kernelspecs", "7.0"),
166 }
168 # Method copied from
169 # https://github.com/jupyterhub/jupyterhub/blob/d1a85e53dccfc7b1dd81b0c1985d158cc6b61820/jupyterhub/auth.py#L143-L161
170 @observe(*list(_deprecated_aliases))
171 def _deprecated_trait(self, change):
172 """observer for deprecated traits"""
173 old_attr = change.name
174 new_attr, version = self._deprecated_aliases[old_attr]
175 new_value = getattr(self, new_attr)
176 if new_value != change.new:
177 # only warn if different
178 # protects backward-compatible config from warnings
179 # if they set the same value under both names
180 self.log.warning(
181 (
182 "{cls}.{old} is deprecated in jupyter_client "
183 "{version}, use {cls}.{new} instead"
184 ).format(
185 cls=self.__class__.__name__,
186 old=old_attr,
187 new=new_attr,
188 version=version,
189 )
190 )
191 setattr(self, new_attr, change.new)
193 def _kernel_dirs_default(self):
194 dirs = jupyter_path("kernels")
195 # At some point, we should stop adding .ipython/kernels to the path,
196 # but the cost to keeping it is very small.
197 try:
198 # this should always be valid on IPython 3+
199 from IPython.paths import get_ipython_dir
201 dirs.append(os.path.join(get_ipython_dir(), "kernels"))
202 except ModuleNotFoundError:
203 pass
204 return dirs
206 def find_kernel_specs(self):
207 """Returns a dict mapping kernel names to resource directories."""
208 d = {}
209 for kernel_dir in self.kernel_dirs:
210 kernels = _list_kernels_in(kernel_dir)
211 for kname, spec in kernels.items():
212 if kname not in d:
213 self.log.debug("Found kernel %s in %s", kname, kernel_dir)
214 d[kname] = spec
216 if self.ensure_native_kernel and NATIVE_KERNEL_NAME not in d:
217 try:
218 from ipykernel.kernelspec import RESOURCES
220 self.log.debug(
221 "Native kernel (%s) available from %s",
222 NATIVE_KERNEL_NAME,
223 RESOURCES,
224 )
225 d[NATIVE_KERNEL_NAME] = RESOURCES
226 except ImportError:
227 self.log.warning("Native kernel (%s) is not available", NATIVE_KERNEL_NAME)
229 if self.allowed_kernelspecs:
230 # filter if there's an allow list
231 d = {name: spec for name, spec in d.items() if name in self.allowed_kernelspecs}
232 return d
233 # TODO: Caching?
235 def _get_kernel_spec_by_name(self, kernel_name, resource_dir):
236 """Returns a :class:`KernelSpec` instance for a given kernel_name
237 and resource_dir.
238 """
239 kspec = None
240 if kernel_name == NATIVE_KERNEL_NAME:
241 try:
242 from ipykernel.kernelspec import RESOURCES, get_kernel_dict
243 except ImportError:
244 # It should be impossible to reach this, but let's play it safe
245 pass
246 else:
247 if resource_dir == RESOURCES:
248 kspec = self.kernel_spec_class(resource_dir=resource_dir, **get_kernel_dict())
249 if not kspec:
250 kspec = self.kernel_spec_class.from_resource_dir(resource_dir)
252 if not KPF.instance(parent=self.parent).is_provisioner_available(kspec):
253 raise NoSuchKernel(kernel_name)
255 return kspec
257 def _find_spec_directory(self, kernel_name):
258 """Find the resource directory of a named kernel spec"""
259 for kernel_dir in [kd for kd in self.kernel_dirs if os.path.isdir(kd)]:
260 files = os.listdir(kernel_dir)
261 for f in files:
262 path = pjoin(kernel_dir, f)
263 if f.lower() == kernel_name and _is_kernel_dir(path):
264 return path
266 if kernel_name == NATIVE_KERNEL_NAME:
267 try:
268 from ipykernel.kernelspec import RESOURCES
269 except ImportError:
270 pass
271 else:
272 return RESOURCES
274 def get_kernel_spec(self, kernel_name):
275 """Returns a :class:`KernelSpec` instance for the given kernel_name.
277 Raises :exc:`NoSuchKernel` if the given kernel name is not found.
278 """
279 if not _is_valid_kernel_name(kernel_name):
280 self.log.warning(
281 f"Kernelspec name {kernel_name} is invalid: {_kernel_name_description}"
282 )
284 resource_dir = self._find_spec_directory(kernel_name.lower())
285 if resource_dir is None:
286 self.log.warning("Kernelspec name %s cannot be found!", kernel_name)
287 raise NoSuchKernel(kernel_name)
289 return self._get_kernel_spec_by_name(kernel_name, resource_dir)
291 def get_all_specs(self):
292 """Returns a dict mapping kernel names to kernelspecs.
294 Returns a dict of the form::
296 {
297 'kernel_name': {
298 'resource_dir': '/path/to/kernel_name',
299 'spec': {"the spec itself": ...}
300 },
301 ...
302 }
303 """
304 d = self.find_kernel_specs()
305 res = {}
306 for kname, resource_dir in d.items():
307 try:
308 if self.__class__ is KernelSpecManager:
309 spec = self._get_kernel_spec_by_name(kname, resource_dir)
310 else:
311 # avoid calling private methods in subclasses,
312 # which may have overridden find_kernel_specs
313 # and get_kernel_spec, but not the newer get_all_specs
314 spec = self.get_kernel_spec(kname)
316 res[kname] = {"resource_dir": resource_dir, "spec": spec.to_dict()}
317 except NoSuchKernel:
318 pass # The appropriate warning has already been logged
319 except Exception:
320 self.log.warning("Error loading kernelspec %r", kname, exc_info=True)
321 return res
323 def remove_kernel_spec(self, name):
324 """Remove a kernel spec directory by name.
326 Returns the path that was deleted.
327 """
328 save_native = self.ensure_native_kernel
329 try:
330 self.ensure_native_kernel = False
331 specs = self.find_kernel_specs()
332 finally:
333 self.ensure_native_kernel = save_native
334 spec_dir = specs[name]
335 self.log.debug("Removing %s", spec_dir)
336 if os.path.islink(spec_dir):
337 os.remove(spec_dir)
338 else:
339 shutil.rmtree(spec_dir)
340 return spec_dir
342 def _get_destination_dir(self, kernel_name, user=False, prefix=None):
343 if user:
344 return os.path.join(self.user_kernel_dir, kernel_name)
345 elif prefix:
346 return os.path.join(os.path.abspath(prefix), "share", "jupyter", "kernels", kernel_name)
347 else:
348 return os.path.join(SYSTEM_JUPYTER_PATH[0], "kernels", kernel_name)
350 def install_kernel_spec(
351 self, source_dir, kernel_name=None, user=False, replace=None, prefix=None
352 ):
353 """Install a kernel spec by copying its directory.
355 If ``kernel_name`` is not given, the basename of ``source_dir`` will
356 be used.
358 If ``user`` is False, it will attempt to install into the systemwide
359 kernel registry. If the process does not have appropriate permissions,
360 an :exc:`OSError` will be raised.
362 If ``prefix`` is given, the kernelspec will be installed to
363 PREFIX/share/jupyter/kernels/KERNEL_NAME. This can be sys.prefix
364 for installation inside virtual or conda envs.
365 """
366 source_dir = source_dir.rstrip("/\\")
367 if not kernel_name:
368 kernel_name = os.path.basename(source_dir)
369 kernel_name = kernel_name.lower()
370 if not _is_valid_kernel_name(kernel_name):
371 msg = f"Invalid kernel name {kernel_name!r}. {_kernel_name_description}"
372 raise ValueError(msg)
374 if user and prefix:
375 msg = "Can't specify both user and prefix. Please choose one or the other."
376 raise ValueError(msg)
378 if replace is not None:
379 warnings.warn(
380 "replace is ignored. Installing a kernelspec always replaces an existing "
381 "installation",
382 DeprecationWarning,
383 stacklevel=2,
384 )
386 destination = self._get_destination_dir(kernel_name, user=user, prefix=prefix)
387 self.log.debug("Installing kernelspec in %s", destination)
389 kernel_dir = os.path.dirname(destination)
390 if kernel_dir not in self.kernel_dirs:
391 self.log.warning(
392 "Installing to %s, which is not in %s. The kernelspec may not be found.",
393 kernel_dir,
394 self.kernel_dirs,
395 )
397 if os.path.isdir(destination):
398 self.log.info("Removing existing kernelspec in %s", destination)
399 shutil.rmtree(destination)
401 shutil.copytree(source_dir, destination)
402 self.log.info("Installed kernelspec %s in %s", kernel_name, destination)
403 return destination
405 def install_native_kernel_spec(self, user=False):
406 """DEPRECATED: Use ipykernel.kernelspec.install"""
407 warnings.warn(
408 "install_native_kernel_spec is deprecated. Use ipykernel.kernelspec import install.",
409 stacklevel=2,
410 )
411 from ipykernel.kernelspec import install
413 install(self, user=user)
416def find_kernel_specs():
417 """Returns a dict mapping kernel names to resource directories."""
418 return KernelSpecManager().find_kernel_specs()
421def get_kernel_spec(kernel_name):
422 """Returns a :class:`KernelSpec` instance for the given kernel_name.
424 Raises KeyError if the given kernel name is not found.
425 """
426 return KernelSpecManager().get_kernel_spec(kernel_name)
429def install_kernel_spec(source_dir, kernel_name=None, user=False, replace=False, prefix=None):
430 """Install a kernel spec in a given directory."""
431 return KernelSpecManager().install_kernel_spec(source_dir, kernel_name, user, replace, prefix)
434install_kernel_spec.__doc__ = KernelSpecManager.install_kernel_spec.__doc__
437def install_native_kernel_spec(user=False):
438 """Install the native kernel spec."""
439 return KernelSpecManager().install_native_kernel_spec(user=user)
442install_native_kernel_spec.__doc__ = KernelSpecManager.install_native_kernel_spec.__doc__