Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/wrapt/importer.py: 23%
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 collections.abc import Callable
10from importlib.util import find_spec
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(".")
37 # Use importlib.import_module() rather than __import__() as the
38 # latter fails for a module registered in sys.modules whose parent
39 # package is not importable. See resolve_path() in patches.py.
41 callback = importlib.import_module(module_name)
42 for attr in attrs:
43 callback = getattr(callback, attr)
44 return callback(module)
46 return import_hook
49def register_post_import_hook(hook, name):
50 """
51 Register a post import hook for the target module `name`. The `hook`
52 function will be called once the module is imported and will be passed the
53 module as argument. If the module is already imported, the `hook` will be
54 called immediately. If you also want to defer loading of the module containing
55 the `hook` function until required, you can specify the `hook` as a string in
56 the form 'module:function'. This will result in a proxy hook function being
57 registered which will defer loading of the specified module containing the
58 callback function until required.
59 """
61 # Create a deferred import hook if hook is a string name rather than
62 # a callable function.
64 if isinstance(hook, str):
65 hook = _create_import_hook_from_string(hook)
67 with _post_import_hooks_lock:
68 # Automatically install the import hook finder if it has not already
69 # been installed.
71 global _post_import_hooks_init
73 if not _post_import_hooks_init:
74 _post_import_hooks_init = True
75 sys.meta_path.insert(0, ImportHookFinder())
77 # Check if the module is already imported. If not, register the hook
78 # to be called after import.
80 module = sys.modules.get(name, None)
82 if module is None:
83 _post_import_hooks.setdefault(name, []).append(hook)
85 # If the module is already imported, we fire the hook right away. Note that
86 # the hook is called outside of the lock to avoid deadlocks if code run as a
87 # consequence of calling the module import hook in turn triggers a separate
88 # thread which tries to register an import hook.
90 if module is not None:
91 hook(module)
94# Register post import hooks defined as package entry points.
97def discover_post_import_hooks(group):
98 """
99 Discover and register post import hooks defined as package entry points
100 in the specified `group`. The group should be a string that matches the
101 entry point group name used in the package metadata.
102 """
104 try:
105 # Python 3.10+ style with select parameter
106 entrypoints = importlib.metadata.entry_points(group=group)
107 except TypeError:
108 # Python 3.8-3.9 style that returns a dict
109 entrypoints = importlib.metadata.entry_points().get(group, ())
111 for entrypoint in entrypoints:
112 callback = entrypoint.load() # Use the loaded callback directly
113 register_post_import_hook(callback, entrypoint.name)
116# Indicate that a module has been loaded. Any post import hooks which
117# were registered against the target module will be invoked. If an
118# exception is raised in any of the post import hooks, that will cause
119# the import of the target module to fail.
122def notify_module_loaded(module):
123 """
124 Notify that a `module` has been loaded and invoke any post import hooks
125 registered against the module. If the module is not registered, this
126 function does nothing.
127 """
129 name = getattr(module, "__name__", None)
131 with _post_import_hooks_lock:
132 hooks = _post_import_hooks.pop(name, ())
134 # Note that the hook is called outside of the lock to avoid deadlocks if
135 # code run as a consequence of calling the module import hook in turn
136 # triggers a separate thread which tries to register an import hook.
138 for hook in hooks:
139 hook(module)
142# A custom module import finder. This intercepts attempts to import
143# modules and watches out for attempts to import target modules of
144# interest. When a module of interest is imported, then any post import
145# hooks which are registered will be invoked.
148class _ImportHookChainedLoader(BaseObjectProxy):
150 def __init__(self, loader):
151 # Explicit class in super() is used because the proxy overrides
152 # __class__ and MRO-related methods to delegate to the wrapped
153 # object, which can interfere with bare super().
154 super(_ImportHookChainedLoader, self).__init__(loader)
156 if hasattr(loader, "load_module"):
157 self.__self_setattr__("load_module", self._self_load_module)
158 if hasattr(loader, "create_module"):
159 self.__self_setattr__("create_module", self._self_create_module)
160 if hasattr(loader, "exec_module"):
161 self.__self_setattr__("exec_module", self._self_exec_module)
163 def _self_set_loader(self, module):
164 # Set module's loader to self.__wrapped__ unless it's already set to
165 # something else. Import machinery will set it to spec.loader if it is
166 # None, so handle None as well. The module may not support attribute
167 # assignment, in which case we simply skip it. Note that we also deal
168 # with __loader__ not existing at all. This is to future proof things
169 # due to proposal to remove the attribute as described in the GitHub
170 # issue at https://github.com/python/cpython/issues/77458. Also prior
171 # to Python 3.3, the __loader__ attribute was only set if a custom
172 # module loader was used. It isn't clear whether the attribute still
173 # existed in that case or was set to None.
175 class UNDEFINED:
176 pass
178 if getattr(module, "__loader__", UNDEFINED) in (None, self):
179 try:
180 module.__loader__ = self.__wrapped__
181 except AttributeError:
182 pass
184 if (
185 getattr(module, "__spec__", None) is not None
186 and getattr(module.__spec__, "loader", None) is self
187 ):
188 module.__spec__.loader = self.__wrapped__
190 def _self_load_module(self, fullname):
191 module = self.__wrapped__.load_module(fullname)
192 self._self_set_loader(module)
193 notify_module_loaded(module)
195 return module
197 # Python 3.4 introduced create_module() and exec_module() instead of
198 # load_module() alone. Splitting the two steps.
200 def _self_create_module(self, spec):
201 return self.__wrapped__.create_module(spec)
203 def _self_exec_module(self, module):
204 self._self_set_loader(module)
205 self.__wrapped__.exec_module(module)
206 notify_module_loaded(module)
209class ImportHookFinder:
211 def __init__(self):
212 self.in_progress = {}
214 def find_module(self, fullname, path=None):
215 with _post_import_hooks_lock:
216 # If the module being imported is not one we have registered
217 # post import hooks for, we can return immediately. We will
218 # take no further part in the importing of this module.
220 if fullname not in _post_import_hooks:
221 return None
223 # When we are interested in a specific module, we will call back
224 # into the import system a second time to defer to the import
225 # finder that is supposed to handle the importing of the module.
226 # We set an in progress flag for the target module so that on
227 # the second time through we don't trigger another call back
228 # into the import system and cause a infinite loop.
230 if fullname in self.in_progress:
231 return None
233 self.in_progress[fullname] = True
235 # Now call back into the import system again.
237 try:
238 # For Python 3 we need to use find_spec().loader
239 # from the importlib.util module. It doesn't actually
240 # import the target module and only finds the
241 # loader. If a loader is found, we need to return
242 # our own loader which will then in turn call the
243 # real loader to import the module and invoke the
244 # post import hooks.
246 loader = getattr(find_spec(fullname), "loader", None)
248 if loader and not isinstance(loader, _ImportHookChainedLoader):
249 return _ImportHookChainedLoader(loader)
251 finally:
252 del self.in_progress[fullname]
254 def find_spec(self, fullname, path=None, target=None):
255 # Since Python 3.4, you are meant to implement find_spec() method
256 # instead of find_module() and since Python 3.10 you get deprecation
257 # warnings if you don't define find_spec().
259 with _post_import_hooks_lock:
260 # If the module being imported is not one we have registered
261 # post import hooks for, we can return immediately. We will
262 # take no further part in the importing of this module.
264 if fullname not in _post_import_hooks:
265 return None
267 # When we are interested in a specific module, we will call back
268 # into the import system a second time to defer to the import
269 # finder that is supposed to handle the importing of the module.
270 # We set an in progress flag for the target module so that on
271 # the second time through we don't trigger another call back
272 # into the import system and cause a infinite loop.
274 if fullname in self.in_progress:
275 return None
277 self.in_progress[fullname] = True
279 # Now call back into the import system again.
281 try:
282 # This should only be Python 3 so find_spec() should always
283 # exist so don't need to check.
285 spec = find_spec(fullname)
286 loader = getattr(spec, "loader", None)
288 if loader and not isinstance(loader, _ImportHookChainedLoader):
289 spec.loader = _ImportHookChainedLoader(loader)
291 return spec
293 finally:
294 del self.in_progress[fullname]
297# Decorator for marking that a function should be called as a post
298# import hook when the target module is imported.
301def when_imported(name):
302 """
303 Returns a decorator that registers the decorated function as a post import
304 hook for the module specified by `name`. The function will be called once
305 the module with the specified name is imported, and will be passed the
306 module as argument. If the module is already imported, the function will
307 be called immediately.
308 """
310 def register(hook):
311 register_post_import_hook(hook, name)
312 return hook
314 return register