import builtins
import sys
from importlib import import_module
from . import LOG
[docs]
class ErrorDuringImport(Exception):
"""Errors that occurred while trying to import something"""
def __init__(self, filename, exc_info):
self.filename = filename
self.exc, self.value, self.tb = exc_info
def __str__(self):
exc = self.exc.__name__
return "problem in %s - %s: %s" % (self.filename, exc, self.value)
[docs]
def safeimport(path, forceload=0, cache=None):
"""Import a module; handle errors; return None if the module isn't found.
If the module *is* found but an exception occurs, it's wrapped in an
ErrorDuringImport exception and reraised. Unlike __import__, if a
package path is specified, the module at the end of the path is
returned, not the package at the beginning. If the optional
'forceload' argument is 1, we reload the module from disk
(unless it's a dynamic extension).
"""
if cache is None:
cache = dict()
LOG.debug("Trying to import '%s'" % path)
try:
# If forceload is 1 and the module has been previously loaded from
# disk, we always have to reload the module. Checking the file's
# mtime isn't good enough (e.g. the module could contain a class
# that inherits from another module that has changed).
if forceload and path in sys.modules:
if path not in sys.builtin_module_names:
# Remove the module from sys.modules and re-import to try
# and avoid problems with partially loaded modules.
# Also remove any submodules because they won't appear
# in the newly loaded module's namespace if they're
# already in sys.modules.
subs = [m for m in sys.modules if m.startswith(path + ".")]
for key in [path] + subs:
# Prevent garbage collection.
cache[key] = sys.modules[key]
del sys.modules[key]
module = __import__(path)
except Exception:
# Did the error occur before or after the module was found?
(exc, value, tb) = info = sys.exc_info()
if path in sys.modules:
# An error occurred while executing the imported module.
raise ErrorDuringImport(sys.modules[path].__file__, info)
elif exc is SyntaxError:
# A SyntaxError occurred before we could execute the module.
raise ErrorDuringImport(value.filename, info)
elif exc is ImportError and value.name == path:
# No such module in the path.
return None
else:
# Some other error occurred during the importing process.
raise ErrorDuringImport(path, sys.exc_info())
for part in path.split(".")[1:]:
try:
module = getattr(module, part)
except AttributeError:
return None
return module
[docs]
def locate(path, forceload=0):
"""Locate an object by name or dotted path, importing as necessary."""
parts = [part for part in path.split(".") if part]
module, n = None, 0
while n < len(parts) - 1:
nextmodule = safeimport(".".join(parts[: n + 1]), forceload)
if nextmodule:
module, n = nextmodule, n + 1
else:
break
if module:
object = module
else:
object = builtins
for part in parts[n:]:
try:
object = getattr(object, part)
except AttributeError as err:
LOG.error("Could not load '%s': %s" % (part, err))
return None
return object
[docs]
def load_with_params(module_path, params):
"""Load a class from *module_path* and pass *params*."""
module, clazz = module_path.split(":")
module = import_module(module)
return getattr(module, clazz)(**params)