Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/wrapt/importer.py: 22%
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"""This module implements a post import hook mechanism styled after what is
2described in PEP-369. Note that it doesn't cope with modules being reloaded.
4"""
6import importlib.metadata
7import sys
8import threading
9from importlib.util import find_spec
10from typing import Callable, Dict, List
12from .__wrapt__ import BaseObjectProxy
14# The dictionary registering any post import hooks to be triggered once
15# the target module has been imported. Once a module has been imported
16# and the hooks fired, the list of hooks recorded against the target
17# module will be truncated but the list left in the dictionary. This
18# acts as a flag to indicate that the module had already been imported.
20_post_import_hooks: Dict[str, List[Callable]] = {}
21_post_import_hooks_init = False
22_post_import_hooks_lock = threading.RLock()
24# Register a new post import hook for the target module name. This
25# differs from the PEP-369 implementation in that it also allows the
26# hook function to be specified as a string consisting of the name of
27# the callback in the form 'module:function'. This will result in a
28# proxy callback being registered which will defer loading of the
29# specified module containing the callback function until required.
32def _create_import_hook_from_string(name):
33 def import_hook(module):
34 module_name, function = name.split(":")
35 attrs = function.split(".")
36 __import__(module_name)
37 callback = sys.modules[module_name]
38 for attr in attrs:
39 callback = getattr(callback, attr)
40 return callback(module)
42 return import_hook
45def register_post_import_hook(hook, name):
46 """
47 Register a post import hook for the target module `name`. The `hook`
48 function will be called once the module is imported and will be passed the
49 module as argument. If the module is already imported, the `hook` will be
50 called immediately. If you also want to defer loading of the module containing
51 the `hook` function until required, you can specify the `hook` as a string in
52 the form 'module:function'. This will result in a proxy hook function being
53 registered which will defer loading of the specified module containing the
54 callback function until required.
55 """
57 # Create a deferred import hook if hook is a string name rather than
58 # a callable function.
60 if isinstance(hook, str):
61 hook = _create_import_hook_from_string(hook)
63 with _post_import_hooks_lock:
64 # Automatically install the import hook finder if it has not already
65 # been installed.
67 global _post_import_hooks_init
69 if not _post_import_hooks_init:
70 _post_import_hooks_init = True
71 sys.meta_path.insert(0, ImportHookFinder())
73 # Check if the module is already imported. If not, register the hook
74 # to be called after import.
76 module = sys.modules.get(name, None)
78 if module is None:
79 _post_import_hooks.setdefault(name, []).append(hook)
81 # If the module is already imported, we fire the hook right away. Note that
82 # the hook is called outside of the lock to avoid deadlocks if code run as a
83 # consequence of calling the module import hook in turn triggers a separate
84 # thread which tries to register an import hook.
86 if module is not None:
87 hook(module)
90# Register post import hooks defined as package entry points.
93def _create_import_hook_from_entrypoint(entrypoint):
94 def import_hook(module):
95 entrypoint_value = entrypoint.value.split(":")
96 module_name = entrypoint_value[0]
97 __import__(module_name)
98 callback = sys.modules[module_name]
100 if len(entrypoint_value) > 1:
101 attrs = entrypoint_value[1].split(".")
102 for attr in attrs:
103 callback = getattr(callback, attr)
104 return callback(module)
106 return import_hook
109def discover_post_import_hooks(group):
110 """
111 Discover and register post import hooks defined as package entry points
112 in the specified `group`. The group should be a string that matches the
113 entry point group name used in the package metadata.
114 """
116 try:
117 # Python 3.10+ style with select parameter
118 entrypoints = importlib.metadata.entry_points(group=group)
119 except TypeError:
120 # Python 3.8-3.9 style that returns a dict
121 entrypoints = importlib.metadata.entry_points().get(group, ())
123 for entrypoint in entrypoints:
124 callback = entrypoint.load() # Use the loaded callback directly
125 register_post_import_hook(callback, entrypoint.name)
128# Indicate that a module has been loaded. Any post import hooks which
129# were registered against the target module will be invoked. If an
130# exception is raised in any of the post import hooks, that will cause
131# the import of the target module to fail.
134def notify_module_loaded(module):
135 """
136 Notify that a `module` has been loaded and invoke any post import hooks
137 registered against the module. If the module is not registered, this
138 function does nothing.
139 """
141 name = getattr(module, "__name__", None)
143 with _post_import_hooks_lock:
144 hooks = _post_import_hooks.pop(name, ())
146 # Note that the hook is called outside of the lock to avoid deadlocks if
147 # code run as a consequence of calling the module import hook in turn
148 # triggers a separate thread which tries to register an import hook.
150 for hook in hooks:
151 hook(module)
154# A custom module import finder. This intercepts attempts to import
155# modules and watches out for attempts to import target modules of
156# interest. When a module of interest is imported, then any post import
157# hooks which are registered will be invoked.
160class _ImportHookLoader:
162 def load_module(self, fullname):
163 module = sys.modules[fullname]
164 notify_module_loaded(module)
166 return module
169class _ImportHookChainedLoader(BaseObjectProxy):
171 def __init__(self, loader):
172 super(_ImportHookChainedLoader, self).__init__(loader)
174 if hasattr(loader, "load_module"):
175 self.__self_setattr__("load_module", self._self_load_module)
176 if hasattr(loader, "create_module"):
177 self.__self_setattr__("create_module", self._self_create_module)
178 if hasattr(loader, "exec_module"):
179 self.__self_setattr__("exec_module", self._self_exec_module)
181 def _self_set_loader(self, module):
182 # Set module's loader to self.__wrapped__ unless it's already set to
183 # something else. Import machinery will set it to spec.loader if it is
184 # None, so handle None as well. The module may not support attribute
185 # assignment, in which case we simply skip it. Note that we also deal
186 # with __loader__ not existing at all. This is to future proof things
187 # due to proposal to remove the attribute as described in the GitHub
188 # issue at https://github.com/python/cpython/issues/77458. Also prior
189 # to Python 3.3, the __loader__ attribute was only set if a custom
190 # module loader was used. It isn't clear whether the attribute still
191 # existed in that case or was set to None.
193 class UNDEFINED:
194 pass
196 if getattr(module, "__loader__", UNDEFINED) in (None, self):
197 try:
198 module.__loader__ = self.__wrapped__
199 except AttributeError:
200 pass
202 if (
203 getattr(module, "__spec__", None) is not None
204 and getattr(module.__spec__, "loader", None) is self
205 ):
206 module.__spec__.loader = self.__wrapped__
208 def _self_load_module(self, fullname):
209 module = self.__wrapped__.load_module(fullname)
210 self._self_set_loader(module)
211 notify_module_loaded(module)
213 return module
215 # Python 3.4 introduced create_module() and exec_module() instead of
216 # load_module() alone. Splitting the two steps.
218 def _self_create_module(self, spec):
219 return self.__wrapped__.create_module(spec)
221 def _self_exec_module(self, module):
222 self._self_set_loader(module)
223 self.__wrapped__.exec_module(module)
224 notify_module_loaded(module)
227class ImportHookFinder:
229 def __init__(self):
230 self.in_progress = {}
232 def find_module(self, fullname, path=None):
233 # If the module being imported is not one we have registered
234 # post import hooks for, we can return immediately. We will
235 # take no further part in the importing of this module.
237 with _post_import_hooks_lock:
238 if fullname not in _post_import_hooks:
239 return None
241 # When we are interested in a specific module, we will call back
242 # into the import system a second time to defer to the import
243 # finder that is supposed to handle the importing of the module.
244 # We set an in progress flag for the target module so that on
245 # the second time through we don't trigger another call back
246 # into the import system and cause a infinite loop.
248 if fullname in self.in_progress:
249 return None
251 self.in_progress[fullname] = True
253 # Now call back into the import system again.
255 try:
256 # For Python 3 we need to use find_spec().loader
257 # from the importlib.util module. It doesn't actually
258 # import the target module and only finds the
259 # loader. If a loader is found, we need to return
260 # our own loader which will then in turn call the
261 # real loader to import the module and invoke the
262 # post import hooks.
264 loader = getattr(find_spec(fullname), "loader", None)
266 if loader and not isinstance(loader, _ImportHookChainedLoader):
267 return _ImportHookChainedLoader(loader)
269 finally:
270 del self.in_progress[fullname]
272 def find_spec(self, fullname, path=None, target=None):
273 # Since Python 3.4, you are meant to implement find_spec() method
274 # instead of find_module() and since Python 3.10 you get deprecation
275 # warnings if you don't define find_spec().
277 # If the module being imported is not one we have registered
278 # post import hooks for, we can return immediately. We will
279 # take no further part in the importing of this module.
281 with _post_import_hooks_lock:
282 if fullname not in _post_import_hooks:
283 return None
285 # When we are interested in a specific module, we will call back
286 # into the import system a second time to defer to the import
287 # finder that is supposed to handle the importing of the module.
288 # We set an in progress flag for the target module so that on
289 # the second time through we don't trigger another call back
290 # into the import system and cause a infinite loop.
292 if fullname in self.in_progress:
293 return None
295 self.in_progress[fullname] = True
297 # Now call back into the import system again.
299 try:
300 # This should only be Python 3 so find_spec() should always
301 # exist so don't need to check.
303 spec = find_spec(fullname)
304 loader = getattr(spec, "loader", None)
306 if loader and not isinstance(loader, _ImportHookChainedLoader):
307 spec.loader = _ImportHookChainedLoader(loader)
309 return spec
311 finally:
312 del self.in_progress[fullname]
315# Decorator for marking that a function should be called as a post
316# import hook when the target module is imported.
319def when_imported(name):
320 """
321 Returns a decorator that registers the decorated function as a post import
322 hook for the module specified by `name`. The function will be called once
323 the module with the specified name is imported, and will be passed the
324 module as argument. If the module is already imported, the function will
325 be called immediately.
326 """
328 def register(hook):
329 register_post_import_hook(hook, name)
330 return hook
332 return register