Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/wrapt/importer.py: 22%
127 statements
« prev ^ index » next coverage.py v7.4.0, created at 2024-01-03 07:57 +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
18from .__wrapt__ import ObjectProxy
20# The dictionary registering any post import hooks to be triggered once
21# the target module has been imported. Once a module has been imported
22# and the hooks fired, the list of hooks recorded against the target
23# module will be truncated but the list left in the dictionary. This
24# acts as a flag to indicate that the module had already been imported.
26_post_import_hooks = {}
27_post_import_hooks_init = False
28_post_import_hooks_lock = threading.RLock()
30# Register a new post import hook for the target module name. This
31# differs from the PEP-369 implementation in that it also allows the
32# hook function to be specified as a string consisting of the name of
33# the callback in the form 'module:function'. This will result in a
34# proxy callback being registered which will defer loading of the
35# specified module containing the callback function until required.
37def _create_import_hook_from_string(name):
38 def import_hook(module):
39 module_name, function = name.split(':')
40 attrs = function.split('.')
41 __import__(module_name)
42 callback = sys.modules[module_name]
43 for attr in attrs:
44 callback = getattr(callback, attr)
45 return callback(module)
46 return import_hook
48def register_post_import_hook(hook, name):
49 # Create a deferred import hook if hook is a string name rather than
50 # a callable function.
52 if isinstance(hook, string_types):
53 hook = _create_import_hook_from_string(hook)
55 with _post_import_hooks_lock:
56 # Automatically install the import hook finder if it has not already
57 # been installed.
59 global _post_import_hooks_init
61 if not _post_import_hooks_init:
62 _post_import_hooks_init = True
63 sys.meta_path.insert(0, ImportHookFinder())
65 # Check if the module is already imported. If not, register the hook
66 # to be called after import.
68 module = sys.modules.get(name, None)
70 if module is None:
71 _post_import_hooks.setdefault(name, []).append(hook)
73 # If the module is already imported, we fire the hook right away. Note that
74 # the hook is called outside of the lock to avoid deadlocks if code run as a
75 # consequence of calling the module import hook in turn triggers a separate
76 # thread which tries to register an import hook.
78 if module is not None:
79 hook(module)
81# Register post import hooks defined as package entry points.
83def _create_import_hook_from_entrypoint(entrypoint):
84 def import_hook(module):
85 __import__(entrypoint.module_name)
86 callback = sys.modules[entrypoint.module_name]
87 for attr in entrypoint.attrs:
88 callback = getattr(callback, attr)
89 return callback(module)
90 return import_hook
92def discover_post_import_hooks(group):
93 try:
94 import pkg_resources
95 except ImportError:
96 return
98 for entrypoint in pkg_resources.iter_entry_points(group=group):
99 callback = _create_import_hook_from_entrypoint(entrypoint)
100 register_post_import_hook(callback, entrypoint.name)
102# Indicate that a module has been loaded. Any post import hooks which
103# were registered against the target module will be invoked. If an
104# exception is raised in any of the post import hooks, that will cause
105# the import of the target module to fail.
107def notify_module_loaded(module):
108 name = getattr(module, '__name__', None)
110 with _post_import_hooks_lock:
111 hooks = _post_import_hooks.pop(name, ())
113 # Note that the hook is called outside of the lock to avoid deadlocks if
114 # code run as a consequence of calling the module import hook in turn
115 # triggers a separate thread which tries to register an import hook.
117 for hook in hooks:
118 hook(module)
120# A custom module import finder. This intercepts attempts to import
121# modules and watches out for attempts to import target modules of
122# interest. When a module of interest is imported, then any post import
123# hooks which are registered will be invoked.
125class _ImportHookLoader:
127 def load_module(self, fullname):
128 module = sys.modules[fullname]
129 notify_module_loaded(module)
131 return module
133class _ImportHookChainedLoader(ObjectProxy):
135 def __init__(self, loader):
136 super(_ImportHookChainedLoader, self).__init__(loader)
138 if hasattr(loader, "load_module"):
139 self.__self_setattr__('load_module', self._self_load_module)
140 if hasattr(loader, "create_module"):
141 self.__self_setattr__('create_module', self._self_create_module)
142 if hasattr(loader, "exec_module"):
143 self.__self_setattr__('exec_module', self._self_exec_module)
145 def _self_set_loader(self, module):
146 # Set module's loader to self.__wrapped__ unless it's already set to
147 # something else. Import machinery will set it to spec.loader if it is
148 # None, so handle None as well. The module may not support attribute
149 # assignment, in which case we simply skip it. Note that we also deal
150 # with __loader__ not existing at all. This is to future proof things
151 # due to proposal to remove the attribue as described in the GitHub
152 # issue at https://github.com/python/cpython/issues/77458. Also prior
153 # to Python 3.3, the __loader__ attribute was only set if a custom
154 # module loader was used. It isn't clear whether the attribute still
155 # existed in that case or was set to None.
157 class UNDEFINED: pass
159 if getattr(module, "__loader__", UNDEFINED) in (None, self):
160 try:
161 module.__loader__ = self.__wrapped__
162 except AttributeError:
163 pass
165 if (getattr(module, "__spec__", None) is not None
166 and getattr(module.__spec__, "loader", None) is self):
167 module.__spec__.loader = self.__wrapped__
169 def _self_load_module(self, fullname):
170 module = self.__wrapped__.load_module(fullname)
171 self._self_set_loader(module)
172 notify_module_loaded(module)
174 return module
176 # Python 3.4 introduced create_module() and exec_module() instead of
177 # load_module() alone. Splitting the two steps.
179 def _self_create_module(self, spec):
180 return self.__wrapped__.create_module(spec)
182 def _self_exec_module(self, module):
183 self._self_set_loader(module)
184 self.__wrapped__.exec_module(module)
185 notify_module_loaded(module)
187class ImportHookFinder:
189 def __init__(self):
190 self.in_progress = {}
192 def find_module(self, fullname, path=None):
193 # If the module being imported is not one we have registered
194 # post import hooks for, we can return immediately. We will
195 # take no further part in the importing of this module.
197 with _post_import_hooks_lock:
198 if fullname not in _post_import_hooks:
199 return None
201 # When we are interested in a specific module, we will call back
202 # into the import system a second time to defer to the import
203 # finder that is supposed to handle the importing of the module.
204 # We set an in progress flag for the target module so that on
205 # the second time through we don't trigger another call back
206 # into the import system and cause a infinite loop.
208 if fullname in self.in_progress:
209 return None
211 self.in_progress[fullname] = True
213 # Now call back into the import system again.
215 try:
216 if not find_spec:
217 # For Python 2 we don't have much choice but to
218 # call back in to __import__(). This will
219 # actually cause the module to be imported. If no
220 # module could be found then ImportError will be
221 # raised. Otherwise we return a loader which
222 # returns the already loaded module and invokes
223 # the post import hooks.
225 __import__(fullname)
227 return _ImportHookLoader()
229 else:
230 # For Python 3 we need to use find_spec().loader
231 # from the importlib.util module. It doesn't actually
232 # import the target module and only finds the
233 # loader. If a loader is found, we need to return
234 # our own loader which will then in turn call the
235 # real loader to import the module and invoke the
236 # post import hooks.
238 loader = getattr(find_spec(fullname), "loader", None)
240 if loader and not isinstance(loader, _ImportHookChainedLoader):
241 return _ImportHookChainedLoader(loader)
243 finally:
244 del self.in_progress[fullname]
246 def find_spec(self, fullname, path=None, target=None):
247 # Since Python 3.4, you are meant to implement find_spec() method
248 # instead of find_module() and since Python 3.10 you get deprecation
249 # warnings if you don't define find_spec().
251 # If the module being imported is not one we have registered
252 # post import hooks for, we can return immediately. We will
253 # take no further part in the importing of this module.
255 with _post_import_hooks_lock:
256 if fullname not in _post_import_hooks:
257 return None
259 # When we are interested in a specific module, we will call back
260 # into the import system a second time to defer to the import
261 # finder that is supposed to handle the importing of the module.
262 # We set an in progress flag for the target module so that on
263 # the second time through we don't trigger another call back
264 # into the import system and cause a infinite loop.
266 if fullname in self.in_progress:
267 return None
269 self.in_progress[fullname] = True
271 # Now call back into the import system again.
273 try:
274 # This should only be Python 3 so find_spec() should always
275 # exist so don't need to check.
277 spec = find_spec(fullname)
278 loader = getattr(spec, "loader", None)
280 if loader and not isinstance(loader, _ImportHookChainedLoader):
281 spec.loader = _ImportHookChainedLoader(loader)
283 return spec
285 finally:
286 del self.in_progress[fullname]
288# Decorator for marking that a function should be called as a post
289# import hook when the target module is imported.
291def when_imported(name):
292 def register(hook):
293 register_post_import_hook(hook, name)
294 return hook
295 return register