Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/jupyter_client/provisioning/factory.py: 31%
70 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-07-01 06:54 +0000
1"""Kernel Provisioner Classes"""
2# Copyright (c) Jupyter Development Team.
3# Distributed under the terms of the Modified BSD License.
4import glob
5import sys
6from os import getenv, path
7from typing import Any, Dict, List
9# See compatibility note on `group` keyword in https://docs.python.org/3/library/importlib.metadata.html#entry-points
10if sys.version_info < (3, 10): # pragma: no cover
11 from importlib_metadata import EntryPoint, entry_points
12else: # pragma: no cover
13 from importlib.metadata import EntryPoint, entry_points
15from traitlets.config import SingletonConfigurable, Unicode, default
17from .provisioner_base import KernelProvisionerBase
20class KernelProvisionerFactory(SingletonConfigurable):
21 """
22 :class:`KernelProvisionerFactory` is responsible for creating provisioner instances.
24 A singleton instance, `KernelProvisionerFactory` is also used by the :class:`KernelSpecManager`
25 to validate `kernel_provisioner` references found in kernel specifications to confirm their
26 availability (in cases where the kernel specification references a kernel provisioner that has
27 not been installed into the current Python environment).
29 It's ``default_provisioner_name`` attribute can be used to specify the default provisioner
30 to use when a kernel_spec is found to not reference a provisioner. It's value defaults to
31 `"local-provisioner"` which identifies the local provisioner implemented by
32 :class:`LocalProvisioner`.
33 """
35 GROUP_NAME = 'jupyter_client.kernel_provisioners'
36 provisioners: Dict[str, EntryPoint] = {}
38 default_provisioner_name_env = "JUPYTER_DEFAULT_PROVISIONER_NAME"
39 default_provisioner_name = Unicode(
40 config=True,
41 help="""Indicates the name of the provisioner to use when no kernel_provisioner
42 entry is present in the kernelspec.""",
43 )
45 @default('default_provisioner_name')
46 def _default_provisioner_name_default(self):
47 """The default provisioner name."""
48 return getenv(self.default_provisioner_name_env, "local-provisioner")
50 def __init__(self, **kwargs: Any) -> None:
51 """Initialize a kernel provisioner factory."""
52 super().__init__(**kwargs)
54 for ep in KernelProvisionerFactory._get_all_provisioners():
55 self.provisioners[ep.name] = ep
57 def is_provisioner_available(self, kernel_spec: Any) -> bool:
58 """
59 Reads the associated ``kernel_spec`` to determine the provisioner and returns whether it
60 exists as an entry_point (True) or not (False). If the referenced provisioner is not
61 in the current cache or cannot be loaded via entry_points, a warning message is issued
62 indicating it is not available.
63 """
64 is_available: bool = True
65 provisioner_cfg = self._get_provisioner_config(kernel_spec)
66 provisioner_name = str(provisioner_cfg.get('provisioner_name'))
67 if not self._check_availability(provisioner_name):
68 is_available = False
69 self.log.warning(
70 f"Kernel '{kernel_spec.display_name}' is referencing a kernel "
71 f"provisioner ('{provisioner_name}') that is not available. "
72 f"Ensure the appropriate package has been installed and retry."
73 )
74 return is_available
76 def create_provisioner_instance(
77 self, kernel_id: str, kernel_spec: Any, parent: Any
78 ) -> KernelProvisionerBase:
79 """
80 Reads the associated ``kernel_spec`` to see if it has a `kernel_provisioner` stanza.
81 If one exists, it instantiates an instance. If a kernel provisioner is not
82 specified in the kernel specification, a default provisioner stanza is fabricated
83 and instantiated corresponding to the current value of ``default_provisioner_name`` trait.
84 The instantiated instance is returned.
86 If the provisioner is found to not exist (not registered via entry_points),
87 `ModuleNotFoundError` is raised.
88 """
89 provisioner_cfg = self._get_provisioner_config(kernel_spec)
90 provisioner_name = str(provisioner_cfg.get('provisioner_name'))
91 if not self._check_availability(provisioner_name):
92 msg = f"Kernel provisioner '{provisioner_name}' has not been registered."
93 raise ModuleNotFoundError(msg)
95 self.log.debug(
96 f"Instantiating kernel '{kernel_spec.display_name}' with "
97 f"kernel provisioner: {provisioner_name}"
98 )
99 provisioner_class = self.provisioners[provisioner_name].load()
100 provisioner_config = provisioner_cfg.get('config')
101 provisioner: KernelProvisionerBase = provisioner_class(
102 kernel_id=kernel_id, kernel_spec=kernel_spec, parent=parent, **provisioner_config
103 )
104 return provisioner
106 def _check_availability(self, provisioner_name: str) -> bool:
107 """
108 Checks that the given provisioner is available.
110 If the given provisioner is not in the current set of loaded provisioners an attempt
111 is made to fetch the named entry point and, if successful, loads it into the cache.
113 :param provisioner_name:
114 :return:
115 """
116 is_available = True
117 if provisioner_name not in self.provisioners:
118 try:
119 ep = self._get_provisioner(provisioner_name)
120 self.provisioners[provisioner_name] = ep # Update cache
121 except Exception:
122 is_available = False
123 return is_available
125 def _get_provisioner_config(self, kernel_spec: Any) -> Dict[str, Any]:
126 """
127 Return the kernel_provisioner stanza from the kernel_spec.
129 Checks the kernel_spec's metadata dictionary for a kernel_provisioner entry.
130 If found, it is returned, else one is created relative to the DEFAULT_PROVISIONER
131 and returned.
133 Parameters
134 ----------
135 kernel_spec : Any - this is a KernelSpec type but listed as Any to avoid circular import
136 The kernel specification object from which the provisioner dictionary is derived.
138 Returns
139 -------
140 dict
141 The provisioner portion of the kernel_spec. If one does not exist, it will contain
142 the default information. If no `config` sub-dictionary exists, an empty `config`
143 dictionary will be added.
144 """
145 env_provisioner = kernel_spec.metadata.get('kernel_provisioner', {})
146 if 'provisioner_name' in env_provisioner: # If no provisioner_name, return default
147 if (
148 'config' not in env_provisioner
149 ): # if provisioner_name, but no config stanza, add one
150 env_provisioner.update({"config": {}})
151 return env_provisioner # Return what we found (plus config stanza if necessary)
152 return {"provisioner_name": self.default_provisioner_name, "config": {}}
154 def get_provisioner_entries(self) -> Dict[str, str]:
155 """
156 Returns a dictionary of provisioner entries.
158 The key is the provisioner name for its entry point. The value is the colon-separated
159 string of the entry point's module name and object name.
160 """
161 entries = {}
162 for name, ep in self.provisioners.items():
163 entries[name] = ep.value
164 return entries
166 @staticmethod
167 def _get_all_provisioners() -> List[EntryPoint]:
168 """Wrapper around entry_points (to fetch the set of provisioners) - primarily to facilitate testing."""
169 return entry_points(group=KernelProvisionerFactory.GROUP_NAME)
171 def _get_provisioner(self, name: str) -> EntryPoint:
172 """Wrapper around entry_points (to fetch a single provisioner) - primarily to facilitate testing."""
173 eps = entry_points(group=KernelProvisionerFactory.GROUP_NAME, name=name)
174 if eps:
175 return eps[0]
177 # Check if the entrypoint name is 'local-provisioner'. Although this should never
178 # happen, we have seen cases where the previous distribution of jupyter_client has
179 # remained which doesn't include kernel-provisioner entrypoints (so 'local-provisioner'
180 # is deemed not found even though its definition is in THIS package). In such cases,
181 # the entrypoints package uses what it first finds - which is the older distribution
182 # resulting in a violation of a supposed invariant condition. To address this scenario,
183 # we will log a warning message indicating this situation, then build the entrypoint
184 # instance ourselves - since we have that information.
185 if name == 'local-provisioner':
186 distros = glob.glob(f"{path.dirname(path.dirname(__file__))}-*")
187 self.log.warning(
188 f"Kernel Provisioning: The 'local-provisioner' is not found. This is likely "
189 f"due to the presence of multiple jupyter_client distributions and a previous "
190 f"distribution is being used as the source for entrypoints - which does not "
191 f"include 'local-provisioner'. That distribution should be removed such that "
192 f"only the version-appropriate distribution remains (version >= 7). Until "
193 f"then, a 'local-provisioner' entrypoint will be automatically constructed "
194 f"and used.\nThe candidate distribution locations are: {distros}"
195 )
196 return EntryPoint(
197 'local-provisioner', 'jupyter_client.provisioning', 'LocalProvisioner'
198 )
200 raise