"""
"""
import json
from galaxy.util.plugin_config import plugins_dict
DEFAULT_SHELL_PLUGIN = "LocalShell"
ERROR_MESSAGE_NO_JOB_PLUGIN = "No job plugin parameter found, cannot create CLI job interface"
ERROR_MESSAGE_NO_SUCH_JOB_PLUGIN = "Failed to find job_plugin of type %s, available types include %s"
[docs]class CliInterface:
"""
High-level interface for loading shell and job plugins and matching
them to specified parameters.
"""
[docs] def __init__(self):
""" """
module_prefix = self.__module__
self.cli_shells = plugins_dict(f"{module_prefix}.shell", "__name__")
self.cli_job_interfaces = plugins_dict(f"{module_prefix}.job", "__name__")
self.active_cli_shells = {}
[docs] def get_plugins(self, shell_params, job_params):
"""
Return shell and job interface defined by and configured via
specified params.
"""
shell = self.get_shell_plugin(shell_params)
job_interface = self.get_job_interface(job_params)
return shell, job_interface
[docs] def get_shell_plugin(self, shell_params):
shell_plugin = shell_params.get("plugin", DEFAULT_SHELL_PLUGIN)
requested_shell_settings = json.dumps(shell_params, sort_keys=True)
if requested_shell_settings not in self.active_cli_shells:
shell_plugin_class = self.cli_shells.get(shell_plugin)
if not shell_plugin_class:
raise ValueError(
f"Unknown shell_plugin [{shell_plugin}], available plugins are {list(self.cli_shells.keys())}"
)
self.active_cli_shells[requested_shell_settings] = shell_plugin_class(**shell_params)
return self.active_cli_shells[requested_shell_settings]
[docs] def get_job_interface(self, job_params):
job_plugin = job_params.get("plugin")
if not job_plugin:
raise ValueError(ERROR_MESSAGE_NO_JOB_PLUGIN)
job_plugin_class = self.cli_job_interfaces.get(job_plugin)
if not job_plugin_class:
raise ValueError(ERROR_MESSAGE_NO_SUCH_JOB_PLUGIN % (job_plugin, list(self.cli_job_interfaces.keys())))
return job_plugin_class(**job_params)
[docs]def split_params(params):
shell_params = {k.replace("shell_", "", 1): v for k, v in params.items() if k.startswith("shell_")}
job_params = {k.replace("job_", "", 1): v for k, v in params.items() if k.startswith("job_")}
return shell_params, job_params