module = None if spec.loader is not None and hasattr(spec.loader, 'create_module'): # It is assumed 'exec_module' will also be defined on the loader. module = spec.loader.create_module(spec) if module is None: module = ModuleType(spec.name) # The import-related module attributes get set here: _init_module_attrs(spec, module)
if spec.loader is None: # unsupported raise ImportError if spec.origin is None and spec.submodule_search_locations is not None: # namespace package sys.modules[spec.name] = module elif not hasattr(spec.loader, 'exec_module'): module = spec.loader.load_module(spec.name) # Set __loader__ and __package__ if missing. else: sys.modules[spec.name] = module try: spec.loader.exec_module(module) except BaseException: try: del sys.modules[spec.name] except KeyError: pass raise return sys.modules[spec.name]
def find_loader(self, fullname): log.debug('find_loader: %r', fullname) parts = fullname.split('.') basename = parts[-1] # Check link cache if self._links is None: self._links = [] # See discussion self._links = _get_links(self._baseurl)
# Check if it's a package if basename in self._links: log.debug('find_loader: trying package %r', fullname) fullurl = self._baseurl + '/' + basename # Attempt to load the package (which accesses __init__.py) loader = UrlPackageLoader(fullurl) try: loader.load_module(fullname) log.debug('find_loader: package %r loaded', fullname) except ImportError as e: log.debug('find_loader: %r is a namespace package', fullname) loader = None return (loader, [fullurl])
# A normal module filename = basename + '.py' if filename in self._links: log.debug('find_loader: module %r found', fullname) return (self._loader, []) else: log.debug('find_loader: module %r not found', fullname) return (None, [])
def invalidate_caches(self): log.debug('invalidating link cache') self._links = None
# Check path to see if it looks like a URL _url_path_cache = {} def handle_url(path): if path.startswith(('http://', 'https://')): log.debug('Handle path? %s. [Yes]', path) if path in _url_path_cache: finder = _url_path_cache[path] else: finder = UrlPathFinder(path) _url_path_cache[path] = finder return finder else: log.debug('Handle path? %s. [No]', path)
if TYPE_CHECKING: from .engine import Engine, get_app from .core import Graph, node_register from .utils.import_utils import load_register_node from .utils.logging_utils import logger_setup else: import sys
import sys import importlib import os from itertools import chain import re
from . import logger
ModuleType = type(sys)
class _LazyModule(ModuleType):
def __init__(self, name, module_file, import_structure, module_spec=None, extra_objects=None): super().__init__(name) self._modules = set(import_structure.keys()) self._class_to_module = {} for key, values in import_structure.items(): for value in values: self._class_to_module[value] = key self.__all__ = list(import_structure.keys()) + list(chain(*import_structure.values())) self.__file__ = module_file self.__spec__ = module_spec self.__path__ = [os.path.dirname(module_file)] self._objects = {} if extra_objects is None else extra_objects self._name = name self._import_structure = import_structure
def __dir__(self): result = super().__dir__() for attr in self.__all__: if attr not in result: result.append(attr) return result
def __getattr__(self, name: str): if name in self._objects: return self._objects[name] if name in self._modules: value = self._get_module(name) elif name in self._class_to_module.keys(): module = self._get_module(self._class_to_module[name]) value = getattr(module, name) else: raise AttributeError(f"module {self.__name__} has no attribute {name}")
setattr(self, name, value) return value
def _get_module(self, module_name: str): try: return importlib.import_module("." + module_name, self.__name__) except Exception as e: raise RuntimeError( f"Failed to import {self.__name__}.{module_name} because of the following error (look up to see its traceback):\n{e}" ) from e