Python importlib._bootstrap 模块，spec_from_file_location() 实例源码

def getInstalledConfig(installDir, configFile):
    """
    Reads config from the installation directory of Plenum.

    :param installDir: installation directory of Plenum
    :param configFile: name of the configuration file
    :raises: FileNotFoundError
    :return: the configuration as a python object
    """
    configPath = os.path.join(installDir, configFile)
    if not os.path.exists(configPath):
        raise FileNotFoundError("No file found at location {}".
                                format(configPath))
    spec = spec_from_file_location(configFile, configPath)
    config = module_from_spec(spec)
    spec.loader.exec_module(config)
    return config

def get_network_name():
    network_name = 'sandbox'
    old_general_config = os.path.join(old_base_dir, 'indy_config.py')
    spec = spec_from_file_location('old_general_config', old_general_config)
    old_cfg = module_from_spec(spec)
    spec.loader.exec_module(old_cfg)
    if hasattr(old_cfg, 'poolTransactionsFile'):
        network_name = _get_network_from_txn_file_name(old_cfg.poolTransactionsFile)
    elif hasattr(old_cfg, 'domainTransactionsFile'):
        network_name = _get_network_from_txn_file_name(old_cfg.domainTransactionsFile)
    elif hasattr(old_cfg, 'current_env') and old_cfg.current_env != 'test':
        network_name = old_cfg.current_env
    return network_name

def getInstalledConfig(installDir, configFile):
    configPath = os.path.join(installDir, configFile)
    if os.path.exists(configPath):
        spec = spec_from_file_location(configFile, configPath)
        config = module_from_spec(spec)
        spec.loader.exec_module(config)
        return config
    else:
        raise FileNotFoundError("No file found at location {}".
                                format(configPath))

def load_napp(self, username, napp_name):
        """Load a single app.

        Load a single NAPP based on its name.

        Args:
            username (str): NApp username present in napp's path.
            napp_name (str): Name of the NApp to be loaded.

        Raises:
            FileNotFoundError: if napps' main.py is not found.

        """
        if (username, napp_name) in self.napps:
            message = 'NApp %s/%s was already loaded'
            self.log.warning(message, username, napp_name)
        else:
            mod_name = '.'.join(['napps', username, napp_name, 'main'])
            path = os.path.join(self.options.napps, username, napp_name,
                                'main.py')
            napp_spec = spec_from_file_location(mod_name, path)
            napp_module = module_from_spec(napp_spec)
            sys.modules[napp_spec.name] = napp_module
            napp_spec.loader.exec_module(napp_module)
            napp = napp_module.Main(controller=self)

            self.napps[(username, napp_name)] = napp

            # This start method is inherited from the Threading class.
            # It is not directly defined/declared on the KytosNApp class.
            napp.start()
            self.api_server.register_napp_endpoints(napp)

            # pylint: disable=protected-access
            for event, listeners in napp._listeners.items():
                self.events_listeners.setdefault(event, []).extend(listeners)
            # pylint: enable=protected-access

def getInstalledConfig(installDir, configFile):
    configPath = os.path.join(installDir, configFile)
    if os.path.exists(configPath):
        spec = spec_from_file_location(configFile, configPath)
        config = module_from_spec(spec)
        spec.loader.exec_module(config)
        return config
    else:
        raise FileNotFoundError("No file found at location {}".
                                format(configPath))

def loadPlugins(plugins_dir, plugins_to_load=None):
    global pluginsLoaded

    alreadyLoadedPlugins = pluginsLoaded.get(plugins_dir)
    i = 0
    if alreadyLoadedPlugins:
        logger.debug("Plugins {} are already loaded from plugins_dir: {}".format(
            alreadyLoadedPlugins, plugins_dir))
    else:
        logger.debug(
            "Plugin loading started to load plugins from plugins_dir: {}".format(
                plugins_dir))

        if not os.path.exists(plugins_dir):
            os.makedirs(plugins_dir)
            logger.debug("Plugin directory created at: {}".format(
                plugins_dir))

        if plugins_to_load is not None:
            for pluginName in plugins_to_load:
                pluginPath = os.path.expanduser(
                    os.path.join(plugins_dir, pluginName + ".py"))
                try:
                    if os.path.exists(pluginPath):
                        spec = spec_from_file_location(
                            pluginName,
                            pluginPath)
                        plugin = module_from_spec(spec)
                        spec.loader.exec_module(plugin)
                        if plugins_dir in pluginsLoaded:
                            pluginsLoaded[plugins_dir].add(pluginName)
                        else:
                            pluginsLoaded[plugins_dir] = {pluginName}
                        i += 1
                    else:
                        if not pluginsNotFound.get(pluginPath):
                            logger.warning(
                                "Note: Plugin file does not exists: {}. "
                                "Create plugin file if you want to load it" .format(pluginPath), extra={
                                    "cli": False})
                            pluginsNotFound[pluginPath] = "Notified"

                except Exception as ex:
                    # TODO: Is this strategy ok to catch any exception and
                    # just print the error and continue,
                    # or it should fail if there is error in plugin loading
                    logger.warning(
                        "** Error occurred during loading plugin {}: {}"
                        .format(pluginPath, str(ex)))

    logger.debug(
        "Total plugins loaded from plugins_dir {} are : {}".format(plugins_dir, i))
    return i