def _path_importer_cache(cls, path): # from importlib.PathFinder
"""Get the finder for the path entry from sys.path_importer_cache.
If the path entry is not in the cache, find the appropriate finder
and cache it. If no finder is available, store None.
"""
if path == '':
try:
path = os.getcwd()
except FileNotFoundError:
# Don't cache the failure as the cwd can easily change to
# a valid directory later on.
return None
try:
finder = sys.path_importer_cache[path]
except KeyError:
finder = cls._path_hooks(path)
sys.path_importer_cache[path] = finder
return finder
python类path_importer_cache()的实例源码
def find_spec(cls, fullname, path=None, target=None):
"""find the module on sys.path or 'path' based on sys.path_hooks and
sys.path_importer_cache."""
if path is None:
path = sys.path
spec = cls._get_spec(fullname, path, target)
if spec is None:
return None
elif spec.loader is None:
namespace_path = spec.submodule_search_locations
if namespace_path:
# We found at least one namespace path. Return a
# spec which can create the namespace package.
spec.origin = 'namespace'
spec.submodule_search_locations = _NamespacePath(fullname, namespace_path, cls._get_spec)
return spec
else:
return None
else:
return spec
# to match importlib API
def print_importers_of(module):
import sys
import pprint
print('MODULE:'),
pprint.pprint(module)
mod_path = module.__file__.split(os.sep)
print()
print('IMPORTERS USED: {')
for i in range(len(mod_path)-1):
parpath = os.sep.join(mod_path[:len(mod_path) - i])
if parpath in sys.path_importer_cache:
cached_imp = sys.path_importer_cache[parpath]
parpath = parpath.replace(sys.prefix, '...')
print('%s: %r' % (parpath, cached_imp))
print('}')
def activate():
"""Install the path-based import components."""
# We should plug filefinder first to avoid plugging ROSDirectoryFinder, when it is not a ROS thing...
filefinder2.activate()
if ros_path_hook not in sys.path_hooks:
# We need to be before FileFinder to be able to find our '.msg' and '.srv' files without making a namespace package
# Note this must be early in the path_hook list, since we change the logic
# and a namespace package becomes a ros importable package.
sys.path_hooks.insert(filefinder2.get_filefinder_index_in_path_hooks(), ros_path_hook)
if ROSPathFinder not in sys.meta_path:
# adding metahook, before the usual pathfinder, to avoid interferences with python namespace mechanism...
sys.meta_path.insert(filefinder2.get_pathfinder_index_in_meta_hooks(), ROSPathFinder)
# Resetting sys.path_importer_cache
# to support the case where we have an implicit (msg/srv) package inside an already loaded package,
# since we need to replace the default importer.
sys.path_importer_cache.clear()
def test_implicit_hooks(self):
# Test that the implicit path hooks are used.
bad_path = '<path>'
module = '<module>'
assert not os.path.exists(bad_path)
existing_path = tempfile.mkdtemp()
try:
with util.import_state():
nothing = _bootstrap._DefaultPathFinder.find_module(module,
path=[existing_path])
self.assertTrue(nothing is None)
self.assertTrue(existing_path in sys.path_importer_cache)
result = isinstance(sys.path_importer_cache[existing_path],
imp.NullImporter)
self.assertFalse(result)
nothing = _bootstrap._DefaultPathFinder.find_module(module,
path=[bad_path])
self.assertTrue(nothing is None)
self.assertTrue(bad_path in sys.path_importer_cache)
self.assertTrue(isinstance(sys.path_importer_cache[bad_path],
imp.NullImporter))
finally:
os.rmdir(existing_path)
def test_path_importer_cache_has_None(self):
# Test that the default hook is used when sys.path_importer_cache
# contains None for a path.
module = '<test module>'
importer = util.mock_modules(module)
path = '<test path>'
# XXX Not blackbox.
original_hook = _bootstrap._DEFAULT_PATH_HOOK
mock_hook = import_util.mock_path_hook(path, importer=importer)
_bootstrap._DEFAULT_PATH_HOOK = mock_hook
try:
with util.import_state(path_importer_cache={path: None}):
loader = _bootstrap._DefaultPathFinder.find_module(module,
path=[path])
self.assertTrue(loader is importer)
finally:
_bootstrap._DEFAULT_PATH_HOOK = original_hook
def test_parallel_path_hooks(self):
# Here the Finder instance is only used to check concurrent calls
# to path_hook().
finder = Finder()
# In order for our path hook to be called at each import, we need
# to flush the path_importer_cache, which we do by registering a
# dedicated meta_path entry.
flushing_finder = FlushingFinder()
def path_hook(path):
finder.find_module('')
raise ImportError
sys.path_hooks.append(path_hook)
sys.meta_path.append(flushing_finder)
try:
# Flush the cache a first time
flushing_finder.find_module('')
numtests = self.check_parallel_module_init()
self.assertGreater(finder.numcalls, 0)
self.assertEqual(finder.x, finder.numcalls)
finally:
sys.meta_path.remove(flushing_finder)
sys.path_hooks.remove(path_hook)
def _check_importer_for_path(name, path_item):
try:
importer = sys.path_importer_cache[path_item]
except KeyError:
for path_hook in sys.path_hooks:
try:
importer = path_hook(path_item)
break
except ImportError:
pass
else:
importer = None
sys.path_importer_cache.setdefault(path_item, importer)
if importer is None:
try:
return imp.find_module(name, [path_item])
except ImportError:
return None
return importer.find_module(name)
def test_parallel_path_hooks(self):
# Here the Finder instance is only used to check concurrent calls
# to path_hook().
finder = Finder()
# In order for our path hook to be called at each import, we need
# to flush the path_importer_cache, which we do by registering a
# dedicated meta_path entry.
flushing_finder = FlushingFinder()
def path_hook(path):
finder.find_module('')
raise ImportError
sys.path_hooks.insert(0, path_hook)
sys.meta_path.append(flushing_finder)
try:
# Flush the cache a first time
flushing_finder.find_module('')
numtests = self.check_parallel_module_init()
self.assertGreater(finder.numcalls, 0)
self.assertEqual(finder.x, finder.numcalls)
finally:
sys.meta_path.remove(flushing_finder)
sys.path_hooks.remove(path_hook)
def test_method_called(self):
# If defined the method should be called.
class InvalidatingNullFinder:
def __init__(self, *ignored):
self.called = False
def find_module(self, *args):
return None
def invalidate_caches(self):
self.called = True
key = 'gobledeegook'
meta_ins = InvalidatingNullFinder()
path_ins = InvalidatingNullFinder()
sys.meta_path.insert(0, meta_ins)
self.addCleanup(lambda: sys.path_importer_cache.__delitem__(key))
sys.path_importer_cache[key] = path_ins
self.addCleanup(lambda: sys.meta_path.remove(meta_ins))
importlib.invalidate_caches()
self.assertTrue(meta_ins.called)
self.assertTrue(path_ins.called)
def get_importer(path_item):
"""Retrieve a PEP 302 importer for the given path item
The returned importer is cached in sys.path_importer_cache
if it was newly created by a path hook.
The cache (or part of it) can be cleared manually if a
rescan of sys.path_hooks is necessary.
"""
try:
importer = sys.path_importer_cache[path_item]
except KeyError:
for path_hook in sys.path_hooks:
try:
importer = path_hook(path_item)
sys.path_importer_cache.setdefault(path_item, importer)
break
except ImportError:
pass
else:
importer = None
return importer
def test_parallel_path_hooks(self):
# Here the Finder instance is only used to check concurrent calls
# to path_hook().
finder = Finder()
# In order for our path hook to be called at each import, we need
# to flush the path_importer_cache, which we do by registering a
# dedicated meta_path entry.
flushing_finder = FlushingFinder()
def path_hook(path):
finder.find_spec('')
raise ImportError
sys.path_hooks.insert(0, path_hook)
sys.meta_path.append(flushing_finder)
try:
# Flush the cache a first time
flushing_finder.find_spec('')
numtests = self.check_parallel_module_init()
self.assertGreater(finder.numcalls, 0)
self.assertEqual(finder.x, finder.numcalls)
finally:
sys.meta_path.remove(flushing_finder)
sys.path_hooks.remove(path_hook)
def test_method_called(self):
# If defined the method should be called.
class InvalidatingNullFinder:
def __init__(self, *ignored):
self.called = False
def find_module(self, *args):
return None
def invalidate_caches(self):
self.called = True
key = 'gobledeegook'
meta_ins = InvalidatingNullFinder()
path_ins = InvalidatingNullFinder()
sys.meta_path.insert(0, meta_ins)
self.addCleanup(lambda: sys.path_importer_cache.__delitem__(key))
sys.path_importer_cache[key] = path_ins
self.addCleanup(lambda: sys.meta_path.remove(meta_ins))
self.init.invalidate_caches()
self.assertTrue(meta_ins.called)
self.assertTrue(path_ins.called)
def test_None_on_sys_path(self):
# Putting None in sys.path[0] caused an import regression from Python
# 3.2: http://bugs.python.org/issue16514
new_path = sys.path[:]
new_path.insert(0, None)
new_path_importer_cache = sys.path_importer_cache.copy()
new_path_importer_cache.pop(None, None)
new_path_hooks = [zipimport.zipimporter,
self.machinery.FileFinder.path_hook(
*self.importlib._bootstrap._get_supported_file_loaders())]
missing = object()
email = sys.modules.pop('email', missing)
try:
with util.import_state(meta_path=sys.meta_path[:],
path=new_path,
path_importer_cache=new_path_importer_cache,
path_hooks=new_path_hooks):
module = self.importlib.import_module('email')
self.assertIsInstance(module, ModuleType)
finally:
if email is not missing:
sys.modules['email'] = email
def get_importer(path_item):
"""Retrieve a PEP 302 importer for the given path item
The returned importer is cached in sys.path_importer_cache
if it was newly created by a path hook.
The cache (or part of it) can be cleared manually if a
rescan of sys.path_hooks is necessary.
"""
try:
importer = sys.path_importer_cache[path_item]
except KeyError:
for path_hook in sys.path_hooks:
try:
importer = path_hook(path_item)
sys.path_importer_cache.setdefault(path_item, importer)
break
except ImportError:
pass
else:
importer = None
return importer
def test_parallel_path_hooks(self):
# Here the Finder instance is only used to check concurrent calls
# to path_hook().
finder = Finder()
# In order for our path hook to be called at each import, we need
# to flush the path_importer_cache, which we do by registering a
# dedicated meta_path entry.
flushing_finder = FlushingFinder()
def path_hook(path):
finder.find_spec('')
raise ImportError
sys.path_hooks.insert(0, path_hook)
sys.meta_path.append(flushing_finder)
try:
# Flush the cache a first time
flushing_finder.find_spec('')
numtests = self.check_parallel_module_init()
self.assertGreater(finder.numcalls, 0)
self.assertEqual(finder.x, finder.numcalls)
finally:
sys.meta_path.remove(flushing_finder)
sys.path_hooks.remove(path_hook)
def test_None_on_sys_path(self):
# Putting None in sys.path[0] caused an import regression from Python
# 3.2: http://bugs.python.org/issue16514
new_path = sys.path[:]
new_path.insert(0, None)
new_path_importer_cache = sys.path_importer_cache.copy()
new_path_importer_cache.pop(None, None)
new_path_hooks = [zipimport.zipimporter,
self.machinery.FileFinder.path_hook(
*self.importlib._bootstrap._get_supported_file_loaders())]
missing = object()
email = sys.modules.pop('email', missing)
try:
with util.import_state(meta_path=sys.meta_path[:],
path=new_path,
path_importer_cache=new_path_importer_cache,
path_hooks=new_path_hooks):
module = self.importlib.import_module('email')
self.assertIsInstance(module, ModuleType)
finally:
if email is not missing:
sys.modules['email'] = email
def _check_importer_for_path(name, path_item):
try:
importer = sys.path_importer_cache[path_item]
except KeyError:
for path_hook in sys.path_hooks:
try:
importer = path_hook(path_item)
break
except ImportError:
pass
else:
importer = None
sys.path_importer_cache.setdefault(path_item, importer)
if importer is None:
try:
return imp.find_module(name, [path_item])
except ImportError:
return None
return importer.find_module(name)
def _get_importer(path_name):
"""Python version of PyImport_GetImporter C API function"""
cache = sys.path_importer_cache
try:
importer = cache[path_name]
except KeyError:
# Not yet cached. Flag as using the
# standard machinery until we finish
# checking the hooks
cache[path_name] = None
for hook in sys.path_hooks:
try:
importer = hook(path_name)
break
except ImportError:
pass
else:
# The following check looks a bit odd. The trick is that
# NullImporter throws ImportError if the supplied path is a
# *valid* directory entry (and hence able to be handled
# by the standard import machinery)
try:
importer = imp.NullImporter(path_name)
except ImportError:
return None
cache[path_name] = importer
return importer
def get_importer(path_item):
"""Retrieve a PEP 302 importer for the given path item
The returned importer is cached in sys.path_importer_cache
if it was newly created by a path hook.
If there is no importer, a wrapper around the basic import
machinery is returned. This wrapper is never inserted into
the importer cache (None is inserted instead).
The cache (or part of it) can be cleared manually if a
rescan of sys.path_hooks is necessary.
"""
try:
importer = sys.path_importer_cache[path_item]
except KeyError:
for path_hook in sys.path_hooks:
try:
importer = path_hook(path_item)
break
except ImportError:
pass
else:
importer = None
sys.path_importer_cache.setdefault(path_item, importer)
if importer is None:
try:
importer = ImpImporter(path_item)
except ImportError:
importer = None
return importer
def uncache_zipdir(path):
"""Ensure that the importer caches dont have stale info for `path`"""
from zipimport import _zip_directory_cache as zdc
_uncache(path, zdc)
_uncache(path, sys.path_importer_cache)
def _get_importer(path_name):
"""Python version of PyImport_GetImporter C API function"""
cache = sys.path_importer_cache
try:
importer = cache[path_name]
except KeyError:
# Not yet cached. Flag as using the
# standard machinery until we finish
# checking the hooks
cache[path_name] = None
for hook in sys.path_hooks:
try:
importer = hook(path_name)
break
except ImportError:
pass
else:
# The following check looks a bit odd. The trick is that
# NullImporter raises ImportError if the supplied path is a
# *valid* directory entry (and hence able to be handled
# by the standard import machinery)
try:
importer = imp.NullImporter(path_name)
except ImportError:
return None
cache[path_name] = importer
return importer
def _findEntryPathString(self, modobj):
"""
Determine where a given Python module object came from by looking at path
entries.
"""
topPackageObj = modobj
while '.' in topPackageObj.__name__:
topPackageObj = self.moduleDict['.'.join(
topPackageObj.__name__.split('.')[:-1])]
if _isPackagePath(FilePath(topPackageObj.__file__)):
# if package 'foo' is on sys.path at /a/b/foo, package 'foo's
# __file__ will be /a/b/foo/__init__.py, and we are looking for
# /a/b here, the path-entry; so go up two steps.
rval = dirname(dirname(topPackageObj.__file__))
else:
# the module is completely top-level, not within any packages. The
# path entry it's on is just its dirname.
rval = dirname(topPackageObj.__file__)
# There are probably some awful tricks that an importer could pull
# which would break this, so let's just make sure... it's a loaded
# module after all, which means that its path MUST be in
# path_importer_cache according to PEP 302 -glyph
from pprint import pformat
assert rval in self.importerCache, '%r for %r not in import cache %s' % (
rval, modobj, pformat(self.importerCache))
return rval
def get_importer(path_item):
"""Retrieve a PEP 302 importer for the given path item
The returned importer is cached in sys.path_importer_cache
if it was newly created by a path hook.
If there is no importer, a wrapper around the basic import
machinery is returned. This wrapper is never inserted into
the importer cache (None is inserted instead).
The cache (or part of it) can be cleared manually if a
rescan of sys.path_hooks is necessary.
"""
try:
importer = sys.path_importer_cache[path_item]
except KeyError:
for path_hook in sys.path_hooks:
try:
importer = path_hook(path_item)
break
except ImportError:
pass
else:
importer = None
sys.path_importer_cache.setdefault(path_item, importer)
if importer is None:
try:
importer = ImpImporter(path_item)
except ImportError:
importer = None
return importer
def checkpath_iter(self, fullname):
for dirpath in sys.path:
# if the value in sys.path_importer_cache is None, then this
# path *should* be imported by the builtin mechanism, and the
# entry is thus a path to a directory on the filesystem;
# if it's not None, then some other importer is in charge, and
# it probably isn't even a filesystem path
if sys.path_importer_cache.get(dirpath,False) is None:
checkpath = os.path.join(
dirpath,'%s.%s' % (fullname,self.suffix))
yield checkpath
def verify_requirements():
sys.path_importer_cache = {} # I don't know if the cache reset has any
basic = find_spec("discord") # side effect. Without it, the lib folder
audio = find_spec("nacl") # wouldn't be seen if it didn't exist
if not basic: # when the launcher was started
return None
elif not audio:
return False
else:
return True
def invalidate_caches(cls):
"""Call the invalidate_caches() method on all path entry finders
stored in sys.path_importer_caches (where implemented)."""
for finder in sys.path_importer_cache.values():
if hasattr(finder, 'invalidate_caches'):
finder.invalidate_caches()
def find_module(cls, fullname, path=None):
"""find the module on sys.path or 'path' based on sys.path_hooks and
sys.path_importer_cache.
This method is for python2 only
"""
spec = cls.find_spec(fullname, path)
if spec is None:
return None
elif spec.loader is None and spec.submodule_search_locations:
# Here we need to create a namespace loader to handle namespaces since python2 doesn't...
return NamespaceLoader2(spec.name, spec.submodule_search_locations)
else:
return spec.loader
def activate():
"""Install the path-based import components."""
if path_hook not in sys.path_hooks:
sys.path_hooks.append(path_hook)
# Resetting sys.path_importer_cache values,
# to support the case where we have an implicit package inside an already loaded package,
# since we need to replace the default importer.
sys.path_importer_cache.clear()
if PathFinder2 not in sys.meta_path:
# Setting up the meta_path to change package finding logic
sys.meta_path.append(PathFinder2)
def get_importer(path_item):
"""Retrieve a PEP 302 importer for the given path item
The returned importer is cached in sys.path_importer_cache
if it was newly created by a path hook.
If there is no importer, a wrapper around the basic import
machinery is returned. This wrapper is never inserted into
the importer cache (None is inserted instead).
The cache (or part of it) can be cleared manually if a
rescan of sys.path_hooks is necessary.
"""
try:
importer = sys.path_importer_cache[path_item]
except KeyError:
for path_hook in sys.path_hooks:
try:
importer = path_hook(path_item)
break
except ImportError:
pass
else:
importer = None
sys.path_importer_cache.setdefault(path_item, importer)
if importer is None:
try:
importer = ImpImporter(path_item)
except ImportError:
importer = None
return importer