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: Activity, arguments: list = []) → PluginChecks

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

Parameters:

msg (Activity) – 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

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.