ss-tools/docs/plugin_dev.md

Plugin Development Guide

This guide explains how to create new plugins for the Superset Tools application.

1. Plugin Structure

A plugin is a single Python file located in the backend/src/plugins/ directory. Each plugin file must contain a class that inherits from PluginBase.

2. Implementing PluginBase

The PluginBase class is an abstract base class that defines the interface for all plugins. You must implement the following properties and methods:

• id: A unique string identifier for your plugin (e.g., "my-cool-plugin").
• name: A human-readable name for your plugin (e.g., "My Cool Plugin").
• description: A brief description of what your plugin does.
• version: The version of your plugin (e.g., "1.0.0").
• get_schema(): A method that returns a JSON schema dictionary defining the input parameters for your plugin. This schema is used to automatically generate a form in the frontend.
• execute(params: Dict[str, Any]): An async method that contains the main logic of your plugin. The params argument is a dictionary containing the input data from the user, validated against the schema you defined.

3. Example Plugin

Here is an example of a simple "Hello World" plugin:

# backend/src/plugins/hello.py
# [DEF:HelloWorldPlugin:Plugin]
# @SEMANTICS: hello, world, example, plugin
# @PURPOSE: A simple "Hello World" plugin example.
# @LAYER: Domain (Plugin)
# @RELATION: Inherits from PluginBase
# @PUBLIC_API: execute

from typing import Dict, Any
from ..core.plugin_base import PluginBase

class HelloWorldPlugin(PluginBase):
    @property
    def id(self) -> str:
        return "hello-world"

    @property
    def name(self) -> str:
        return "Hello World"

    @property
    def description(self) -> str:
        return "A simple plugin that prints a greeting."

    @property
    def version(self) -> str:
        return "1.0.0"

    def get_schema(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "name": {
                    "type": "string",
                    "title": "Name",
                    "description": "The name to greet.",
                    "default": "World",
                }
            },
            "required": ["name"],
        }

    async def execute(self, params: Dict[str, Any], context: Optional[TaskContext] = None):
        name = params["name"]
        if context:
            context.logger.info(f"Hello, {name}!")
        else:
            print(f"Hello, {name}!")

4. Logging with TaskContext

Plugins now support TaskContext for structured logging with source attribution. The context parameter provides access to a logger that automatically tags logs with the task ID and a source identifier.

4.1. Basic Logging

Use context.logger to log messages with automatic source attribution:

from typing import Dict, Any, Optional
from ..core.plugin_base import PluginBase
from ..core.task_manager.context import TaskContext

async def execute(self, params: Dict[str, Any], context: Optional[TaskContext] = None):
    if context:
        # Use TaskContext logger for structured logging
        context.logger.info("My plugin is running!")
    else:
        # Fallback to global logger for backward compatibility
        from ..core.logger import logger
        logger.info("My plugin is running!")

4.2. Source Attribution

For better log organization, create sub-loggers for different components:

async def execute(self, params: Dict[str, Any], context: Optional[TaskContext] = None):
    if context:
        # Create sub-loggers for different components
        api_log = context.logger.with_source("api")
        storage_log = context.logger.with_source("storage")
        
        api_log.info("Connecting to API...")
        storage_log.info("Saving file...")
    else:
        # Fallback to global logger
        from ..core.logger import logger
        logger.info("My plugin is running!")

4.3. Log Levels

The logger supports standard log levels. Use them appropriately:

Level	Usage
DEBUG	Detailed diagnostic information (API responses, internal state). Only visible when log level is set to DEBUG.
INFO	General operational messages (start/complete notifications, progress updates).
WARNING	Non-critical issues that don't stop execution (deprecated APIs, retry attempts).
ERROR	Failures that prevent an operation from completing (API errors, validation failures).

# Good: Use DEBUG for verbose diagnostic info
api_log.debug(f"API response: {response.json()}")

# Good: Use INFO for operational milestones
log.info(f"Starting backup for environment: {env}")

# Good: Use WARNING for recoverable issues
log.warning(f"Rate limit hit, retrying in {delay}s")

# Good: Use ERROR for failures
log.error(f"Failed to connect to database: {e}")

4.4. Progress Logging

For operations that report progress, use the progress method:

async def execute(self, params: Dict[str, Any], context: Optional[TaskContext] = None):
    if context:
        total_items = 100
        for i, item in enumerate(items):
            # Report progress with percentage
            percent = (i + 1) / total_items * 100
            context.logger.progress(f"Processing {item}", percent=percent)
    else:
        # Fallback
        from ..core.logger import logger
        logger.info("My plugin is running!")

4.5. Logging with Metadata

You can include structured metadata with log entries:

async def execute(self, params: Dict[str, Any], context: Optional[TaskContext] = None):
    if context:
        context.logger.error(
            "Operation failed",
            metadata={"error_code": 500, "details": "Connection timeout"}
        )
    else:
        from ..core.logger import logger
        logger.error("Operation failed")

4.6. Common Source Names

For consistency across plugins, use these standard source names:

Source	Usage
superset_api	Superset REST API calls
postgres	PostgreSQL database operations
storage	File system operations
git	Git operations
llm	LLM API calls
screenshot	Screenshot capture operations
migration	Migration-specific logic
backup	Backup operations
debug	Debug/diagnostic operations
search	Search operations

4.7. Best Practices

1. Always check for context: Support backward compatibility by checking if context is available:

   log = context.logger if context else logger
   

2. Use source attribution: Create sub-loggers for different components to make filtering easier in the UI.

3. Use appropriate log levels:

   • DEBUG: Verbose diagnostic info (API responses, internal state)
   • INFO: Operational milestones (start, complete, progress)
   • WARNING: Recoverable issues (rate limits, deprecated APIs)
   • ERROR: Failures that stop an operation

4. Log progress for long operations: Use progress() for operations that take time:

   for i, item in enumerate(items):
       percent = (i + 1) / len(items) * 100
       log.progress(f"Processing {item}", percent=percent)
   

5. Keep DEBUG logs verbose, INFO logs concise: DEBUG logs can include full API responses, while INFO logs should be one-line summaries.

5. Testing

To test your plugin, simply run the application and navigate to the web UI. Your plugin should appear in the list of available tools.