123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230 |
- """This module implements a post import hook mechanism styled after what is
- described in PEP-369. Note that it doesn't cope with modules being reloaded.
- """
- import sys
- import threading
- PY2 = sys.version_info[0] == 2
- if PY2:
- string_types = basestring,
- else:
- import importlib
- string_types = str,
- from .decorators import synchronized
- # The dictionary registering any post import hooks to be triggered once
- # the target module has been imported. Once a module has been imported
- # and the hooks fired, the list of hooks recorded against the target
- # module will be truncated but the list left in the dictionary. This
- # acts as a flag to indicate that the module had already been imported.
- _post_import_hooks = {}
- _post_import_hooks_init = False
- _post_import_hooks_lock = threading.RLock()
- # Register a new post import hook for the target module name. This
- # differs from the PEP-369 implementation in that it also allows the
- # hook function to be specified as a string consisting of the name of
- # the callback in the form 'module:function'. This will result in a
- # proxy callback being registered which will defer loading of the
- # specified module containing the callback function until required.
- def _create_import_hook_from_string(name):
- def import_hook(module):
- module_name, function = name.split(':')
- attrs = function.split('.')
- __import__(module_name)
- callback = sys.modules[module_name]
- for attr in attrs:
- callback = getattr(callback, attr)
- return callback(module)
- return import_hook
- @synchronized(_post_import_hooks_lock)
- def register_post_import_hook(hook, name):
- # Create a deferred import hook if hook is a string name rather than
- # a callable function.
- if isinstance(hook, string_types):
- hook = _create_import_hook_from_string(hook)
- # Automatically install the import hook finder if it has not already
- # been installed.
- global _post_import_hooks_init
- if not _post_import_hooks_init:
- _post_import_hooks_init = True
- sys.meta_path.insert(0, ImportHookFinder())
- # Determine if any prior registration of a post import hook for
- # the target modules has occurred and act appropriately.
- hooks = _post_import_hooks.get(name, None)
- if hooks is None:
- # No prior registration of post import hooks for the target
- # module. We need to check whether the module has already been
- # imported. If it has we fire the hook immediately and add an
- # empty list to the registry to indicate that the module has
- # already been imported and hooks have fired. Otherwise add
- # the post import hook to the registry.
- module = sys.modules.get(name, None)
- if module is not None:
- _post_import_hooks[name] = []
- hook(module)
- else:
- _post_import_hooks[name] = [hook]
- elif hooks == []:
- # A prior registration of port import hooks for the target
- # module was done and the hooks already fired. Fire the hook
- # immediately.
- module = sys.modules[name]
- hook(module)
- else:
- # A prior registration of port import hooks for the target
- # module was done but the module has not yet been imported.
- _post_import_hooks[name].append(hook)
- # Register post import hooks defined as package entry points.
- def _create_import_hook_from_entrypoint(entrypoint):
- def import_hook(module):
- __import__(entrypoint.module_name)
- callback = sys.modules[entrypoint.module_name]
- for attr in entrypoint.attrs:
- callback = getattr(callback, attr)
- return callback(module)
- return import_hook
- def discover_post_import_hooks(group):
- try:
- import pkg_resources
- except ImportError:
- return
- for entrypoint in pkg_resources.iter_entry_points(group=group):
- callback = _create_import_hook_from_entrypoint(entrypoint)
- register_post_import_hook(callback, entrypoint.name)
- # Indicate that a module has been loaded. Any post import hooks which
- # were registered against the target module will be invoked. If an
- # exception is raised in any of the post import hooks, that will cause
- # the import of the target module to fail.
- @synchronized(_post_import_hooks_lock)
- def notify_module_loaded(module):
- name = getattr(module, '__name__', None)
- hooks = _post_import_hooks.get(name, None)
- if hooks:
- _post_import_hooks[name] = []
- for hook in hooks:
- hook(module)
- # A custom module import finder. This intercepts attempts to import
- # modules and watches out for attempts to import target modules of
- # interest. When a module of interest is imported, then any post import
- # hooks which are registered will be invoked.
- class _ImportHookLoader:
- def load_module(self, fullname):
- module = sys.modules[fullname]
- notify_module_loaded(module)
- return module
- class _ImportHookChainedLoader:
- def __init__(self, loader):
- self.loader = loader
- def load_module(self, fullname):
- module = self.loader.load_module(fullname)
- notify_module_loaded(module)
- return module
- class ImportHookFinder:
- def __init__(self):
- self.in_progress = {}
- @synchronized(_post_import_hooks_lock)
- def find_module(self, fullname, path=None):
- # If the module being imported is not one we have registered
- # post import hooks for, we can return immediately. We will
- # take no further part in the importing of this module.
- if not fullname in _post_import_hooks:
- return None
- # When we are interested in a specific module, we will call back
- # into the import system a second time to defer to the import
- # finder that is supposed to handle the importing of the module.
- # We set an in progress flag for the target module so that on
- # the second time through we don't trigger another call back
- # into the import system and cause a infinite loop.
- if fullname in self.in_progress:
- return None
- self.in_progress[fullname] = True
- # Now call back into the import system again.
- try:
- if PY2:
- # For Python 2 we don't have much choice but to
- # call back in to __import__(). This will
- # actually cause the module to be imported. If no
- # module could be found then ImportError will be
- # raised. Otherwise we return a loader which
- # returns the already loaded module and invokes
- # the post import hooks.
- __import__(fullname)
- return _ImportHookLoader()
- else:
- # For Python 3 we need to use find_spec().loader
- # from the importlib.util module. It doesn't actually
- # import the target module and only finds the
- # loader. If a loader is found, we need to return
- # our own loader which will then in turn call the
- # real loader to import the module and invoke the
- # post import hooks.
- try:
- import importlib.util
- loader = importlib.util.find_spec(fullname).loader
- except (ImportError, AttributeError):
- loader = importlib.find_loader(fullname, path)
- if loader:
- return _ImportHookChainedLoader(loader)
- finally:
- del self.in_progress[fullname]
- # Decorator for marking that a function should be called as a post
- # import hook when the target module is imported.
- def when_imported(name):
- def register(hook):
- register_post_import_hook(hook, name)
- return hook
- return register
|