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: Logger | None = None)

Plugins class takes care of managing all plugins.

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

Parameters:

logger – The logger where to log information/warning or errors.

check(msg: AbstractMessage, arguments: dict | None = None) PluginChecks
check(msg: str, arguments: dict = {}) PluginChecks

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

Parameters:
  • plugin_name (str) – 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.

Return type:

PluginChecks

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

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 of str

Examples

If nothing has been configured

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

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 of 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:

The names of all the plugins that have been registered.

Return type:

list of str

run(msg: AbstractMessage, arguments: dict | None = 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)