Developer API

class zambeze.orchestration.plugins.PluginChecks(val: dict)

error_detected() → bool: Detects if an error was found in the results of the plugin checks

class zambeze.orchestration.plugins.Plugins(logger: Optional[Logger] = None)

Plugins class takes care of managing all plugins.

Plugins can be added as plugins by creating packages in the plugin_modules

Parameters: logger (Optional[logging.Logger]) – The logger where to log information/warning or errors.

check(msg: AbstractMessage, arguments: Optional[dict] = None) → PluginChecks

check(msg: str, arguments: dict = {}) → PluginChecks

Check that the arguments passed to the plugin “plugin_name” are valid

Parameters

plugin_name (str) – the name of the plugin to validate against
arguments (dict) – the arguments to be validated for plugin “plugin_name”

Returns

What is returned are a list of the plugins and their actions along with an indication on whether there was a problem with them

Example

Using rsync

For the rsync plugin to be useful, both the local and remote host ssh keys must have been configured. By default the rsync plugin will look for the private key located at ~/.ssh/id_rsa. If the private key is different then it must be specified with the “private_ssh_key” key value pair.

plugins = Plugins() config = {

“rsync”: {
“private_ssh_key”: “path to private ssh key” }

}

plugins.configure(config) arguments = {

“transfer”: {

“source”: {
“ip”: local_ip, user”: current_user, “path”: current_valid_path, },

“destination”: {
“ip”: “172.22.1.69”, “user”: “cades”, “path”: “/home/cades/josh-testing”, },

“arguments”: [“-a”], }

} checked_args = plugins.check(“rsync”, arguments) print(checked_args) >> { >> “rsync”: { “transfer”: (True, “”) } >> }

configure(config: dict)

Configuration options for each plugin

This method is responsible for initializing all the plugins that are supported in the plugin_modules folder. It should be called before the plugins can be run, all plugins should be configured before they can be run.

Parameters: config (dict) – This contains relevant configuration information for each plugin, if provided will only configure the plugins listed
Example: Arguments

The configuration options for each plugin will appear under their name in the configuration parameter.

I.e. for plugins ‘globus’ and ‘shell’

config = {

‘globus’: {

‘authentication flow’: {: ‘type’: ‘credential flow’, ‘secret’: “blahblah” },
‘shell’: {: ‘arguments’ : [‘’] } }

plugins = Plugins() plugins.configure(config, [‘shell’])

This will just configure the “shell” plugin

property configured: list[str]

Will return a list of all the plugins that have been configured.

Returns: list of all plugins that are ready to be run
Return type: list[str]
Example: If nothing has been configured

plugins = Plugins() assert len(plugins.configured) == 0

Example: If globus is configured

config = {

“globus”: {: “client id”: “…” }

}

plugins.configure(config) assert len(plugins.configured) == 1 assert “globus” in plugins.configured

property info: dict

Will return the current state of the registered plugins

Parameters: plugins (list[str]) – the plugins to provide information about defaults to information about all plugins
Returns: The actual information of each plugin that was specified
Return type: dict
Example

these_plugins = [“globus”, “shell”] plugins.configure(configuration_options) information = plugins.info(these_plugins) print(information)

{: “globus”: {…} “shell”: {…}

}

property registered: list[str]

List all plugins that have been registered.

This method can be called at any time and is meant to simply display which packages are supported and present in the plugin_modules folder. It does not mean that these plugins have been configured. All plugins must be configured before they can be run.

Returns: Returns the names of all the plugins that have been registered
Return type: list[str]

run(msg: AbstractMessage, arguments: Optional[dict] = None) → None

run(msg: str, arguments: dict = {}) → None

Run a specific plugins.

Parameters

plugin_name (str) – Plugin name
arguments (dict) – Plugin arguments

Example

plugins = Plugins() config = {

“rsync”: {
“ssh_key”: “path to private ssh key” }

}

plugins.configure(config) arguments = {

“transfer”: {

“source”: {
“ip”: “hostname”: “path”: },

“destination”: {
“ip”: “hostname”: “path”: }

}

}

# Should return True for each action that was found to be # correctly validated checks = plugins.check(‘rsync’, arguments) print(checks) # Should print { “rsync”: { “transfer”: True } } plugins.run(‘rsync’, arguments)