Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/jupyter_client/provisioning/provisioner_base.py: 54%
69 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 os
5from abc import ABC, ABCMeta, abstractmethod
6from typing import Any, Dict, List, Optional, Union
8from traitlets.config import Instance, LoggingConfigurable, Unicode
10from ..connect import KernelConnectionInfo
13class KernelProvisionerMeta(ABCMeta, type(LoggingConfigurable)): # type: ignore
14 pass
17class KernelProvisionerBase( # type:ignore[misc]
18 ABC, LoggingConfigurable, metaclass=KernelProvisionerMeta
19):
20 """
21 Abstract base class defining methods for KernelProvisioner classes.
23 A majority of methods are abstract (requiring implementations via a subclass) while
24 some are optional and others provide implementations common to all instances.
25 Subclasses should be aware of which methods require a call to the superclass.
27 Many of these methods model those of :class:`subprocess.Popen` for parity with
28 previous versions where the kernel process was managed directly.
29 """
31 # The kernel specification associated with this provisioner
32 kernel_spec: Any = Instance('jupyter_client.kernelspec.KernelSpec', allow_none=True)
33 kernel_id: Union[str, Unicode] = Unicode(None, allow_none=True)
34 connection_info: KernelConnectionInfo = {}
36 @property
37 @abstractmethod
38 def has_process(self) -> bool:
39 """
40 Returns true if this provisioner is currently managing a process.
42 This property is asserted to be True immediately following a call to
43 the provisioner's :meth:`launch_kernel` method.
44 """
45 pass
47 @abstractmethod
48 async def poll(self) -> Optional[int]:
49 """
50 Checks if kernel process is still running.
52 If running, None is returned, otherwise the process's integer-valued exit code is returned.
53 This method is called from :meth:`KernelManager.is_alive`.
54 """
55 pass
57 @abstractmethod
58 async def wait(self) -> Optional[int]:
59 """
60 Waits for kernel process to terminate.
62 This method is called from `KernelManager.finish_shutdown()` and
63 `KernelManager.kill_kernel()` when terminating a kernel gracefully or
64 immediately, respectively.
65 """
66 pass
68 @abstractmethod
69 async def send_signal(self, signum: int) -> None:
70 """
71 Sends signal identified by signum to the kernel process.
73 This method is called from `KernelManager.signal_kernel()` to send the
74 kernel process a signal.
75 """
76 pass
78 @abstractmethod
79 async def kill(self, restart: bool = False) -> None:
80 """
81 Kill the kernel process.
83 This is typically accomplished via a SIGKILL signal, which cannot be caught.
84 This method is called from `KernelManager.kill_kernel()` when terminating
85 a kernel immediately.
87 restart is True if this operation will precede a subsequent launch_kernel request.
88 """
89 pass
91 @abstractmethod
92 async def terminate(self, restart: bool = False) -> None:
93 """
94 Terminates the kernel process.
96 This is typically accomplished via a SIGTERM signal, which can be caught, allowing
97 the kernel provisioner to perform possible cleanup of resources. This method is
98 called indirectly from `KernelManager.finish_shutdown()` during a kernel's
99 graceful termination.
101 restart is True if this operation precedes a start launch_kernel request.
102 """
103 pass
105 @abstractmethod
106 async def launch_kernel(self, cmd: List[str], **kwargs: Any) -> KernelConnectionInfo:
107 """
108 Launch the kernel process and return its connection information.
110 This method is called from `KernelManager.launch_kernel()` during the
111 kernel manager's start kernel sequence.
112 """
113 pass
115 @abstractmethod
116 async def cleanup(self, restart: bool = False) -> None:
117 """
118 Cleanup any resources allocated on behalf of the kernel provisioner.
120 This method is called from `KernelManager.cleanup_resources()` as part of
121 its shutdown kernel sequence.
123 restart is True if this operation precedes a start launch_kernel request.
124 """
125 pass
127 async def shutdown_requested(self, restart: bool = False) -> None:
128 """
129 Allows the provisioner to determine if the kernel's shutdown has been requested.
131 This method is called from `KernelManager.request_shutdown()` as part of
132 its shutdown sequence.
134 This method is optional and is primarily used in scenarios where the provisioner
135 may need to perform other operations in preparation for a kernel's shutdown.
136 """
137 pass
139 async def pre_launch(self, **kwargs: Any) -> Dict[str, Any]:
140 """
141 Perform any steps in preparation for kernel process launch.
143 This includes applying additional substitutions to the kernel launch command
144 and environment. It also includes preparation of launch parameters.
146 NOTE: Subclass implementations are advised to call this method as it applies
147 environment variable substitutions from the local environment and calls the
148 provisioner's :meth:`_finalize_env()` method to allow each provisioner the
149 ability to cleanup the environment variables that will be used by the kernel.
151 This method is called from `KernelManager.pre_start_kernel()` as part of its
152 start kernel sequence.
154 Returns the (potentially updated) keyword arguments that are passed to
155 :meth:`launch_kernel()`.
156 """
157 env = kwargs.pop('env', os.environ).copy()
158 env.update(self.__apply_env_substitutions(env))
159 self._finalize_env(env)
160 kwargs['env'] = env
162 return kwargs
164 async def post_launch(self, **kwargs: Any) -> None:
165 """
166 Perform any steps following the kernel process launch.
168 This method is called from `KernelManager.post_start_kernel()` as part of its
169 start kernel sequence.
170 """
171 pass
173 async def get_provisioner_info(self) -> Dict[str, Any]:
174 """
175 Captures the base information necessary for persistence relative to this instance.
177 This enables applications that subclass `KernelManager` to persist a kernel provisioner's
178 relevant information to accomplish functionality like disaster recovery or high availability
179 by calling this method via the kernel manager's `provisioner` attribute.
181 NOTE: The superclass method must always be called first to ensure proper serialization.
182 """
183 provisioner_info: Dict[str, Any] = {}
184 provisioner_info['kernel_id'] = self.kernel_id
185 provisioner_info['connection_info'] = self.connection_info
186 return provisioner_info
188 async def load_provisioner_info(self, provisioner_info: Dict) -> None:
189 """
190 Loads the base information necessary for persistence relative to this instance.
192 The inverse of `get_provisioner_info()`, this enables applications that subclass
193 `KernelManager` to re-establish communication with a provisioner that is managing
194 a (presumably) remote kernel from an entirely different process that the original
195 provisioner.
197 NOTE: The superclass method must always be called first to ensure proper deserialization.
198 """
199 self.kernel_id = provisioner_info['kernel_id']
200 self.connection_info = provisioner_info['connection_info']
202 def get_shutdown_wait_time(self, recommended: float = 5.0) -> float:
203 """
204 Returns the time allowed for a complete shutdown. This may vary by provisioner.
206 This method is called from `KernelManager.finish_shutdown()` during the graceful
207 phase of its kernel shutdown sequence.
209 The recommended value will typically be what is configured in the kernel manager.
210 """
211 return recommended
213 def get_stable_start_time(self, recommended: float = 10.0) -> float:
214 """
215 Returns the expected upper bound for a kernel (re-)start to complete.
216 This may vary by provisioner.
218 The recommended value will typically be what is configured in the kernel restarter.
219 """
220 return recommended
222 def _finalize_env(self, env: Dict[str, str]) -> None:
223 """
224 Ensures env is appropriate prior to launch.
226 This method is called from `KernelProvisionerBase.pre_launch()` during the kernel's
227 start sequence.
229 NOTE: Subclasses should be sure to call super()._finalize_env(env)
230 """
231 if self.kernel_spec.language and self.kernel_spec.language.lower().startswith("python"):
232 # Don't allow PYTHONEXECUTABLE to be passed to kernel process.
233 # If set, it can bork all the things.
234 env.pop('PYTHONEXECUTABLE', None)
236 def __apply_env_substitutions(self, substitution_values: Dict[str, str]) -> Dict[str, str]:
237 """
238 Walks entries in the kernelspec's env stanza and applies substitutions from current env.
240 This method is called from `KernelProvisionerBase.pre_launch()` during the kernel's
241 start sequence.
243 Returns the substituted list of env entries.
245 NOTE: This method is private and is not intended to be overridden by provisioners.
246 """
247 substituted_env = {}
248 if self.kernel_spec:
249 from string import Template
251 # For each templated env entry, fill any templated references
252 # matching names of env variables with those values and build
253 # new dict with substitutions.
254 templated_env = self.kernel_spec.env
255 for k, v in templated_env.items():
256 substituted_env.update({k: Template(v).safe_substitute(substitution_values)})
257 return substituted_env