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
python类spec_from_file_location()的实例源码
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