Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/wrapt/importer.py: 21%
126 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:35 +0000
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 sys
7import threading
9PY2 = sys.version_info[0] == 2
11if PY2:
12 string_types = basestring,
13 find_spec = None
14else:
15 string_types = str,
16 from importlib.util import find_spec
18# The dictionary registering any post import hooks to be triggered once
19# the target module has been imported. Once a module has been imported
20# and the hooks fired, the list of hooks recorded against the target
21# module will be truncated but the list left in the dictionary. This
22# acts as a flag to indicate that the module had already been imported.
24_post_import_hooks = {}
25_post_import_hooks_init = False
26_post_import_hooks_lock = threading.RLock()
28# Register a new post import hook for the target module name. This
29# differs from the PEP-369 implementation in that it also allows the
30# hook function to be specified as a string consisting of the name of
31# the callback in the form 'module:function'. This will result in a
32# proxy callback being registered which will defer loading of the
33# specified module containing the callback function until required.
35def _create_import_hook_from_string(name):
36 def import_hook(module):
37 module_name, function = name.split(':')
38 attrs = function.split('.')
39 __import__(module_name)
40 callback = sys.modules[module_name]
41 for attr in attrs:
42 callback = getattr(callback, attr)
43 return callback(module)
44 return import_hook
46def register_post_import_hook(hook, name):
47 # Create a deferred import hook if hook is a string name rather than
48 # a callable function.
50 if isinstance(hook, string_types):
51 hook = _create_import_hook_from_string(hook)
53 with _post_import_hooks_lock:
54 # Automatically install the import hook finder if it has not already
55 # been installed.
57 global _post_import_hooks_init
59 if not _post_import_hooks_init:
60 _post_import_hooks_init = True
61 sys.meta_path.insert(0, ImportHookFinder())
63 # Check if the module is already imported. If not, register the hook
64 # to be called after import.
66 module = sys.modules.get(name, None)
68 if module is None:
69 _post_import_hooks.setdefault(name, []).append(hook)
71 # If the module is already imported, we fire the hook right away. Note that
72 # the hook is called outside of the lock to avoid deadlocks if code run as a
73 # consequence of calling the module import hook in turn triggers a separate
74 # thread which tries to register an import hook.
76 if module is not None:
77 hook(module)
79# Register post import hooks defined as package entry points.
81def _create_import_hook_from_entrypoint(entrypoint):
82 def import_hook(module):
83 __import__(entrypoint.module_name)
84 callback = sys.modules[entrypoint.module_name]
85 for attr in entrypoint.attrs:
86 callback = getattr(callback, attr)
87 return callback(module)
88 return import_hook
90def discover_post_import_hooks(group):
91 try:
92 import pkg_resources
93 except ImportError:
94 return
96 for entrypoint in pkg_resources.iter_entry_points(group=group):
97 callback = _create_import_hook_from_entrypoint(entrypoint)
98 register_post_import_hook(callback, entrypoint.name)
100# Indicate that a module has been loaded. Any post import hooks which
101# were registered against the target module will be invoked. If an
102# exception is raised in any of the post import hooks, that will cause
103# the import of the target module to fail.
105def notify_module_loaded(module):
106 name = getattr(module, '__name__', None)
108 with _post_import_hooks_lock:
109 hooks = _post_import_hooks.pop(name, ())
111 # Note that the hook is called outside of the lock to avoid deadlocks if
112 # code run as a consequence of calling the module import hook in turn
113 # triggers a separate thread which tries to register an import hook.
115 for hook in hooks:
116 hook(module)
118# A custom module import finder. This intercepts attempts to import
119# modules and watches out for attempts to import target modules of
120# interest. When a module of interest is imported, then any post import
121# hooks which are registered will be invoked.
123class _ImportHookLoader:
125 def load_module(self, fullname):
126 module = sys.modules[fullname]
127 notify_module_loaded(module)
129 return module
131class _ImportHookChainedLoader:
133 def __init__(self, loader):
134 self.loader = loader
136 if hasattr(loader, "load_module"):
137 self.load_module = self._load_module
138 if hasattr(loader, "create_module"):
139 self.create_module = self._create_module
140 if hasattr(loader, "exec_module"):
141 self.exec_module = self._exec_module
143 def _set_loader(self, module):
144 # Set module's loader to self.loader unless it's already set to
145 # something else. Import machinery will set it to spec.loader if it is
146 # None, so handle None as well. The module may not support attribute
147 # assignment, in which case we simply skip it. Note that we also deal
148 # with __loader__ not existing at all. This is to future proof things
149 # due to proposal to remove the attribue as described in the GitHub
150 # issue at https://github.com/python/cpython/issues/77458. Also prior
151 # to Python 3.3, the __loader__ attribute was only set if a custom
152 # module loader was used. It isn't clear whether the attribute still
153 # existed in that case or was set to None.
155 class UNDEFINED: pass
157 if getattr(module, "__loader__", UNDEFINED) in (None, self):
158 try:
159 module.__loader__ = self.loader
160 except AttributeError:
161 pass
163 if (getattr(module, "__spec__", None) is not None
164 and getattr(module.__spec__, "loader", None) is self):
165 module.__spec__.loader = self.loader
167 def _load_module(self, fullname):
168 module = self.loader.load_module(fullname)
169 self._set_loader(module)
170 notify_module_loaded(module)
172 return module
174 # Python 3.4 introduced create_module() and exec_module() instead of
175 # load_module() alone. Splitting the two steps.
177 def _create_module(self, spec):
178 return self.loader.create_module(spec)
180 def _exec_module(self, module):
181 self._set_loader(module)
182 self.loader.exec_module(module)
183 notify_module_loaded(module)
185class ImportHookFinder:
187 def __init__(self):
188 self.in_progress = {}
190 def find_module(self, fullname, path=None):
191 # If the module being imported is not one we have registered
192 # post import hooks for, we can return immediately. We will
193 # take no further part in the importing of this module.
195 with _post_import_hooks_lock:
196 if fullname not in _post_import_hooks:
197 return None
199 # When we are interested in a specific module, we will call back
200 # into the import system a second time to defer to the import
201 # finder that is supposed to handle the importing of the module.
202 # We set an in progress flag for the target module so that on
203 # the second time through we don't trigger another call back
204 # into the import system and cause a infinite loop.
206 if fullname in self.in_progress:
207 return None
209 self.in_progress[fullname] = True
211 # Now call back into the import system again.
213 try:
214 if not find_spec:
215 # For Python 2 we don't have much choice but to
216 # call back in to __import__(). This will
217 # actually cause the module to be imported. If no
218 # module could be found then ImportError will be
219 # raised. Otherwise we return a loader which
220 # returns the already loaded module and invokes
221 # the post import hooks.
223 __import__(fullname)
225 return _ImportHookLoader()
227 else:
228 # For Python 3 we need to use find_spec().loader
229 # from the importlib.util module. It doesn't actually
230 # import the target module and only finds the
231 # loader. If a loader is found, we need to return
232 # our own loader which will then in turn call the
233 # real loader to import the module and invoke the
234 # post import hooks.
236 loader = getattr(find_spec(fullname), "loader", None)
238 if loader and not isinstance(loader, _ImportHookChainedLoader):
239 return _ImportHookChainedLoader(loader)
241 finally:
242 del self.in_progress[fullname]
244 def find_spec(self, fullname, path=None, target=None):
245 # Since Python 3.4, you are meant to implement find_spec() method
246 # instead of find_module() and since Python 3.10 you get deprecation
247 # warnings if you don't define find_spec().
249 # If the module being imported is not one we have registered
250 # post import hooks for, we can return immediately. We will
251 # take no further part in the importing of this module.
253 with _post_import_hooks_lock:
254 if fullname not in _post_import_hooks:
255 return None
257 # When we are interested in a specific module, we will call back
258 # into the import system a second time to defer to the import
259 # finder that is supposed to handle the importing of the module.
260 # We set an in progress flag for the target module so that on
261 # the second time through we don't trigger another call back
262 # into the import system and cause a infinite loop.
264 if fullname in self.in_progress:
265 return None
267 self.in_progress[fullname] = True
269 # Now call back into the import system again.
271 try:
272 # This should only be Python 3 so find_spec() should always
273 # exist so don't need to check.
275 spec = find_spec(fullname)
276 loader = getattr(spec, "loader", None)
278 if loader and not isinstance(loader, _ImportHookChainedLoader):
279 spec.loader = _ImportHookChainedLoader(loader)
281 return spec
283 finally:
284 del self.in_progress[fullname]
286# Decorator for marking that a function should be called as a post
287# import hook when the target module is imported.
289def when_imported(name):
290 def register(hook):
291 register_post_import_hook(hook, name)
292 return hook
293 return register