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


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

  • msg (Activity) – Name of the plugin to validate against.

  • arguments (dict) – The arguments to be validated for plugin “plugin_name”.


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:


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.


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


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.


List of all plugins that are ready to be run.

Return type:

list of str


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.


plugins (list of str) – The plugins to provide information about defaults to information about all plugins


The actual information of each plugin that was specified.

Return type:



>>> these_plugins = ["globus", "shell"]
>>> plugins.configure(configuration_options)
>>> information =
>>> 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.


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.

  • plugin_name (str) – Plugin name.

  • arguments (dict) – Plugin arguments.