Plugin Operations

Operations for managing Stash plugins.

Bases: StashClientProtocol

Mixin for plugin-related client methods.

Functions

get_plugins async

get_plugins() -> list[Plugin]

Get all loaded plugins.

Returns:

Type	Description
list[Plugin]	List of Plugin objects representing all loaded plugins

Examples:

Get all plugins:

plugins = await client.get_plugins()
for plugin in plugins:
    print(f"Plugin: {plugin.name} (enabled={plugin.enabled})")

Filter enabled plugins:

plugins = await client.get_plugins()
enabled = [p for p in plugins if p.enabled]
print(f"Enabled plugins: {len(enabled)}")

get_plugin_tasks async

get_plugin_tasks() -> list[PluginTask]

Get all available plugin tasks.

Returns:

Type	Description
list[PluginTask]	List of PluginTask objects representing all available plugin operations

Examples:

Get all plugin tasks:

tasks = await client.get_plugin_tasks()
for task in tasks:
    print(f"Task: {task.name} - {task.description}")
    print(f"  Plugin: {task.plugin.name}")

Find tasks for a specific plugin:

tasks = await client.get_plugin_tasks()
plugin_tasks = [t for t in tasks if t.plugin.id == "my-plugin-id"]

set_plugins_enabled async

set_plugins_enabled(
    enabled_map: BoolMap | dict[str, bool],
) -> bool

Enable or disable plugins.

Parameters:

Name	Type	Description	Default
enabled_map	BoolMap | dict[str, bool]	Dictionary mapping plugin IDs to enabled status (True/False). Plugins not in the map are not affected.	required

Returns:

Type	Description
bool	True if the operation was successful, False otherwise

Examples:

Enable a plugin:

success = await client.set_plugins_enabled({
    "my-plugin-id": True
})

Disable multiple plugins:

success = await client.set_plugins_enabled({
    "plugin-1": False,
    "plugin-2": False,
    "plugin-3": True
})

run_plugin_task async

run_plugin_task(
    plugin_id: str,
    task_name: str | None = None,
    description: str | None = None,
    args_map: Map | dict[str, Any] | None = None,
) -> str

Run a plugin task asynchronously via the job queue.

If task_name is provided, the task must exist in the plugin config and the task's configuration will be used. If no task_name is provided, the plugin will be executed with the arguments provided only.

Parameters:

Name	Type	Description	Default
plugin_id	str	ID of the plugin to run	required
task_name	str | None	Optional name of the task to run (uses task's default config)	None
description	str | None	Optional description for the job queue	None
args_map	Map | dict[str, Any] | None	Optional arguments to pass to the plugin (as a dictionary)	None

Returns:

Type	Description
str	Job ID for the plugin task

Raises:

Type	Description
ValueError	If the operation fails or no job ID is returned

Examples:

Run a plugin task with a task name:

job_id = await client.run_plugin_task(
    plugin_id="my-plugin",
    task_name="scan",
    description="Scanning library"
)
await client.wait_for_job(job_id)

Run a plugin with custom arguments:

job_id = await client.run_plugin_task(
    plugin_id="my-plugin",
    args_map={"path": "/videos", "recursive": True}
)

Run with both task name and additional arguments:

job_id = await client.run_plugin_task(
    plugin_id="my-plugin",
    task_name="process",
    description="Processing files",
    args_map={"overwrite": True}
)

run_plugin_operation async

run_plugin_operation(
    plugin_id: str, args: Map | dict[str, Any] | None = None
) -> Any

Run a plugin operation synchronously (not via job queue).

The operation is run immediately and does not use the job queue. Returns the result directly from the plugin.

Parameters:

Name	Type	Description	Default
plugin_id	str	ID of the plugin to run	required
args	Map | dict[str, Any] | None	Optional arguments to pass to the plugin (as a dictionary)	None

Returns:

Type	Description
Any	The result from the plugin operation (type depends on the plugin)

Examples:

Run a plugin operation:

result = await client.run_plugin_operation(
    plugin_id="my-plugin",
    args={"action": "query", "id": "123"}
)
print(f"Plugin returned: {result}")

Run without arguments:

result = await client.run_plugin_operation(plugin_id="my-plugin")

reload_plugins async

reload_plugins() -> bool

Reload all plugins.

Returns:

Type	Description
bool	True if plugins were reloaded successfully, False otherwise

Examples:

Reload all plugins:

success = await client.reload_plugins()
if success:
    print("Plugins reloaded successfully")

configure_plugin async

configure_plugin(
    plugin_id: str, config: Map | dict[str, Any]
) -> Map

Configure a plugin's settings.

Overwrites the entire plugin configuration for the given plugin.

Parameters:

Name	Type	Description	Default
plugin_id	str	ID of the plugin to configure	required
config	Map | dict[str, Any]	Configuration dictionary to set for the plugin	required

Returns:

Type	Description
Map	The updated plugin configuration

Examples:

Configure a plugin:

config = await client.configure_plugin(
    plugin_id="my-plugin",
    config={
        "api_key": "secret-key",
        "endpoint": "https://api.example.com",
        "enabled_features": ["feature1", "feature2"]
    }
)
print(f"Updated config: {config}")

Update specific settings:

# Get current config first
plugins = await client.get_plugins()
my_plugin = next(p for p in plugins if p.id == "my-plugin")

# Modify and save
config = await client.configure_plugin(
    plugin_id="my-plugin",
    config={"setting1": "value1", "setting2": True}
)