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(".")
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 discover_post_import_hooks(group):
94 """
95 Discover and register post import hooks defined as package entry points
96 in the specified `group`. The group should be a string that matches the
97 entry point group name used in the package metadata.
98 """
100 try:
101 # Python 3.10+ style with select parameter
102 entrypoints = importlib.metadata.entry_points(group=group)
103 except TypeError:
104 # Python 3.8-3.9 style that returns a dict
105 entrypoints = importlib.metadata.entry_points().get(group, ())
107 for entrypoint in entrypoints:
108 callback = entrypoint.load() # Use the loaded callback directly
109 register_post_import_hook(callback, entrypoint.name)
112# Indicate that a module has been loaded. Any post import hooks which
113# were registered against the target module will be invoked. If an
114# exception is raised in any of the post import hooks, that will cause
115# the import of the target module to fail.
118def notify_module_loaded(module):
119 """
120 Notify that a `module` has been loaded and invoke any post import hooks
121 registered against the module. If the module is not registered, this
122 function does nothing.
123 """
125 name = getattr(module, "__name__", None)
127 with _post_import_hooks_lock:
128 hooks = _post_import_hooks.pop(name, ())
130 # Note that the hook is called outside of the lock to avoid deadlocks if
131 # code run as a consequence of calling the module import hook in turn
132 # triggers a separate thread which tries to register an import hook.
134 for hook in hooks:
135 hook(module)
138# A custom module import finder. This intercepts attempts to import
139# modules and watches out for attempts to import target modules of
140# interest. When a module of interest is imported, then any post import
141# hooks which are registered will be invoked.
144class _ImportHookChainedLoader(BaseObjectProxy):
146 def __init__(self, loader):
147 # Explicit class in super() is used because the proxy overrides
148 # __class__ and MRO-related methods to delegate to the wrapped
149 # object, which can interfere with bare super().
150 super(_ImportHookChainedLoader, self).__init__(loader)
152 if hasattr(loader, "load_module"):
153 self.__self_setattr__("load_module", self._self_load_module)
154 if hasattr(loader, "create_module"):
155 self.__self_setattr__("create_module", self._self_create_module)
156 if hasattr(loader, "exec_module"):
157 self.__self_setattr__("exec_module", self._self_exec_module)
159 def _self_set_loader(self, module):
160 # Set module's loader to self.__wrapped__ unless it's already set to
161 # something else. Import machinery will set it to spec.loader if it is
162 # None, so handle None as well. The module may not support attribute
163 # assignment, in which case we simply skip it. Note that we also deal
164 # with __loader__ not existing at all. This is to future proof things
165 # due to proposal to remove the attribute as described in the GitHub
166 # issue at https://github.com/python/cpython/issues/77458. Also prior
167 # to Python 3.3, the __loader__ attribute was only set if a custom
168 # module loader was used. It isn't clear whether the attribute still
169 # existed in that case or was set to None.
171 class UNDEFINED:
172 pass
174 if getattr(module, "__loader__", UNDEFINED) in (None, self):
175 try:
176 module.__loader__ = self.__wrapped__
177 except AttributeError:
178 pass
180 if (
181 getattr(module, "__spec__", None) is not None
182 and getattr(module.__spec__, "loader", None) is self
183 ):
184 module.__spec__.loader = self.__wrapped__
186 def _self_load_module(self, fullname):
187 module = self.__wrapped__.load_module(fullname)
188 self._self_set_loader(module)
189 notify_module_loaded(module)
191 return module
193 # Python 3.4 introduced create_module() and exec_module() instead of
194 # load_module() alone. Splitting the two steps.
196 def _self_create_module(self, spec):
197 return self.__wrapped__.create_module(spec)
199 def _self_exec_module(self, module):
200 self._self_set_loader(module)
201 self.__wrapped__.exec_module(module)
202 notify_module_loaded(module)
205class ImportHookFinder:
207 def __init__(self):
208 self.in_progress = {}
210 def find_module(self, fullname, path=None):
211 with _post_import_hooks_lock:
212 # If the module being imported is not one we have registered
213 # post import hooks for, we can return immediately. We will
214 # take no further part in the importing of this module.
216 if fullname not in _post_import_hooks:
217 return None
219 # When we are interested in a specific module, we will call back
220 # into the import system a second time to defer to the import
221 # finder that is supposed to handle the importing of the module.
222 # We set an in progress flag for the target module so that on
223 # the second time through we don't trigger another call back
224 # into the import system and cause a infinite loop.
226 if fullname in self.in_progress:
227 return None
229 self.in_progress[fullname] = True
231 # Now call back into the import system again.
233 try:
234 # For Python 3 we need to use find_spec().loader
235 # from the importlib.util module. It doesn't actually
236 # import the target module and only finds the
237 # loader. If a loader is found, we need to return
238 # our own loader which will then in turn call the
239 # real loader to import the module and invoke the
240 # post import hooks.
242 loader = getattr(find_spec(fullname), "loader", None)
244 if loader and not isinstance(loader, _ImportHookChainedLoader):
245 return _ImportHookChainedLoader(loader)
247 finally:
248 del self.in_progress[fullname]
250 def find_spec(self, fullname, path=None, target=None):
251 # Since Python 3.4, you are meant to implement find_spec() method
252 # instead of find_module() and since Python 3.10 you get deprecation
253 # warnings if you don't define find_spec().
255 with _post_import_hooks_lock:
256 # If the module being imported is not one we have registered
257 # post import hooks for, we can return immediately. We will
258 # take no further part in the importing of this module.
260 if fullname not in _post_import_hooks:
261 return None
263 # When we are interested in a specific module, we will call back
264 # into the import system a second time to defer to the import
265 # finder that is supposed to handle the importing of the module.
266 # We set an in progress flag for the target module so that on
267 # the second time through we don't trigger another call back
268 # into the import system and cause a infinite loop.
270 if fullname in self.in_progress:
271 return None
273 self.in_progress[fullname] = True
275 # Now call back into the import system again.
277 try:
278 # This should only be Python 3 so find_spec() should always
279 # exist so don't need to check.
281 spec = find_spec(fullname)
282 loader = getattr(spec, "loader", None)
284 if loader and not isinstance(loader, _ImportHookChainedLoader):
285 spec.loader = _ImportHookChainedLoader(loader)
287 return spec
289 finally:
290 del self.in_progress[fullname]
293# Decorator for marking that a function should be called as a post
294# import hook when the target module is imported.
297def when_imported(name):
298 """
299 Returns a decorator that registers the decorated function as a post import
300 hook for the module specified by `name`. The function will be called once
301 the module with the specified name is imported, and will be passed the
302 module as argument. If the module is already imported, the function will
303 be called immediately.
304 """
306 def register(hook):
307 register_post_import_hook(hook, name)
308 return hook
310 return register