Creating Plugins¶

from nexus.plugins import NexusPlugin, PluginMetadata
from typing import Callable, Any

class MyPlugin(NexusPlugin):
    """My custom Nexus plugin."""

    def metadata(self) -> PluginMetadata:
        """Return plugin metadata."""
        return PluginMetadata(
            name="my-plugin",
            version="1.0.0",
            description="My custom plugin for Nexus",
            author="Your Name",
            homepage="https://github.com/yourname/nexus-plugin-my",
            requires=[]  # Optional plugin dependencies
        )

    def commands(self) -> dict[str, Callable]:
        """Register CLI commands."""
        return {
            "hello": self.hello_command,
            "list-files": self.list_files_command,
        }

    def hooks(self) -> dict[str, Callable]:
        """Register lifecycle hooks."""
        return {
            "before_write": self.validate_content,
            "after_write": self.log_write,
        }

    async def initialize(self, config: dict[str, Any]) -> None:
        """Initialize plugin with configuration."""
        self._config = config
        # Perform initialization here
        print(f"Initializing {self.metadata().name}...")

    async def shutdown(self) -> None:
        """Cleanup when plugin is disabled."""
        # Cleanup resources here
        print(f"Shutting down {self.metadata().name}...")

    # ===== Command Implementations =====

    async def hello_command(self, name: str = "World"):
        """Say hello to someone.

        Usage:
            nexus my-plugin hello
            nexus my-plugin hello --name Alice
        """
        print(f"Hello, {name}!")

    async def list_files_command(self, path: str = "/"):
        """List files in Nexus.

        Usage:
            nexus my-plugin list-files
            nexus my-plugin list-files --path /workspace
        """
        if not self.nx:
            print("Error: NexusFS not available")
            return

        files = self.nx.list(path, recursive=True)
        print(f"Files in {path}:")
        for file in files:
            print(f"  {file}")

    # ===== Hook Implementations =====

    async def validate_content(self, context: dict) -> dict | None:
        """Validate content before writing."""
        path = context.get("path", "")
        content = context.get("content", b"")

        # Example validation: warn about large files
        if len(content) > 10 * 1024 * 1024:  # 10MB
            print(f"Warning: Large file write: {path} ({len(content)} bytes)")

        return context  # Continue with write

    async def log_write(self, context: dict) -> dict:
        """Log after writing a file."""
        path = context.get("path", "")
        print(f"[my-plugin] File written: {path}")
        return context

from nexus_my_plugin.plugin import MyPlugin

__all__ = ["MyPlugin"]

async def my_command(self, arg1: str, arg2: int = 10):
    """Command with arguments.

    Args:
        arg1: Required string argument
        arg2: Optional integer argument (default: 10)

    Usage:
        nexus my-plugin my-command <arg1>
        nexus my-plugin my-command <arg1> --arg2 20
    """
    print(f"arg1={arg1}, arg2={arg2}")

async def process_file(self, path: str):
    """Process a file using NexusFS.

    Usage:
        nexus my-plugin process-file /data/file.txt
    """
    if not self.nx:
        print("Error: NexusFS not available")
        return

    try:
        # Read file
        content = self.nx.read(path)
        print(f"Processing {path}: {len(content)} bytes")

        # Process content
        processed = content.upper()

        # Write result
        output_path = f"{path}.processed"
        self.nx.write(output_path, processed)
        print(f"Wrote result to {output_path}")

    except Exception as e:
        print(f"Error: {e}")

from rich.console import Console
from rich.table import Table

async def show_stats(self):
    """Show statistics with rich formatting."""
    console = Console()

    if not self.nx:
        console.print("[red]Error: NexusFS not available[/red]")
        return

    # Collect statistics
    files = self.nx.list("/", recursive=True)
    total_size = sum(len(self.nx.read(f)) for f in files)

    # Create table
    table = Table(title="Nexus Statistics")
    table.add_column("Metric", style="cyan")
    table.add_column("Value", style="green")

    table.add_row("Total Files", str(len(files)))
    table.add_row("Total Size", f"{total_size / 1024 / 1024:.2f} MB")

    console.print(table)

async def transform(self):
    """Transform data from pipeline.

    Usage:
        nexus plugin-a output | nexus my-plugin transform | nexus plugin-b input
    """
    if self.is_piped_input():
        # Read JSON from stdin
        try:
            data = self.read_json_input()

            # Transform data
            result = {
                "transformed": True,
                "original": data,
                "timestamp": datetime.now().isoformat()
            }

            # Write JSON to stdout
            self.write_json_output(result)

        except json.JSONDecodeError:
            print("Error: Invalid JSON input", file=sys.stderr)
    else:
        print("Error: This command requires piped input")

async def initialize(self, config: dict[str, Any]) -> None:
    """Initialize with configuration."""
    self._config = config

    # Get configuration values
    api_key = self.get_config("api_key")
    cache_dir = self.get_config("cache_dir", "/tmp/my-plugin")
    enabled_features = self.get_config("features", [])

    # Validate configuration
    if not api_key:
        raise ValueError("api_key is required in configuration")

    # Use configuration
    self._setup_api(api_key)
    self._cache_dir = Path(cache_dir)
    self._cache_dir.mkdir(parents=True, exist_ok=True)

async def my_command(self, path: str):
    """Command with proper error handling."""
    try:
        if not self.nx:
            raise ValueError("NexusFS not available")

        content = self.nx.read(path)
        # Process content...

    except FileNotFoundError:
        print(f"Error: File not found: {path}")
    except PermissionError:
        print(f"Error: Permission denied: {path}")
    except Exception as e:
        print(f"Error: {e}")
        import traceback
        traceback.print_exc()

async def initialize(self, config: dict[str, Any]) -> None:
    """Initialize with configuration validation."""
    self._config = config

    # Validate required config
    required_keys = ["api_key", "endpoint"]
    for key in required_keys:
        if not self.get_config(key):
            raise ValueError(f"{key} is required in plugin configuration")

    # Validate config values
    cache_dir = self.get_config("cache_dir", "/tmp/default")
    if not os.path.exists(cache_dir):
        os.makedirs(cache_dir, exist_ok=True)

async def initialize(self, config: dict[str, Any]) -> None:
    """Initialize resources."""
    self._config = config
    self._client = SomeClient(api_key=config["api_key"])
    self._cache = {}
    print("Initialized my-plugin")

async def shutdown(self) -> None:
    """Cleanup resources."""
    if hasattr(self, "_client"):
        await self._client.close()

    if hasattr(self, "_cache"):
        self._cache.clear()

    print("Shut down my-plugin")

from typing import Callable, Any, Optional

async def my_command(
    self,
    path: str,
    limit: int = 10,
    verbose: bool = False
) -> None:
    """Command with type hints."""
    ...

async def my_command(self, path: str, recursive: bool = False):
    """Process files in a directory.

    This command processes all files in the specified directory
    and generates a report.

    Args:
        path: Directory path to process
        recursive: Whether to process subdirectories (default: False)

    Usage:
        nexus my-plugin my-command /data
        nexus my-plugin my-command /data --recursive

    Examples:
        # Process files in /data directory
        nexus my-plugin my-command /data

        # Process files recursively
        nexus my-plugin my-command /data --recursive
    """
    ...

import pytest
from nexus_my_plugin import MyPlugin
from nexus.plugins import PluginMetadata

@pytest.mark.asyncio
async def test_plugin_metadata():
    """Test plugin metadata."""
    plugin = MyPlugin()
    metadata = plugin.metadata()

    assert isinstance(metadata, PluginMetadata)
    assert metadata.name == "my-plugin"
    assert metadata.version == "1.0.0"

@pytest.mark.asyncio
async def test_hello_command():
    """Test hello command."""
    plugin = MyPlugin()
    # Test command (capture output)
    await plugin.hello_command("Alice")
    # Assert expected behavior

@pytest.mark.asyncio
async def test_with_nexus_fs(tmp_path):
    """Test plugin with NexusFS."""
    from nexus import connect

    # Create test Nexus instance
    nx = connect(config={"data_dir": str(tmp_path)})

    # Create plugin with NexusFS
    plugin = MyPlugin(nx)
    await plugin.initialize({})

    # Test commands
    await plugin.list_files_command("/")

    # Cleanup
    nx.close()

@pytest.mark.asyncio
async def test_plugin_integration(tmp_path):
    """Test plugin with full Nexus integration."""
    from nexus import connect
    from nexus.plugins import PluginRegistry

    # Setup
    nx = connect(config={"data_dir": str(tmp_path)})
    registry = PluginRegistry(nx)

    # Register plugin
    plugin = MyPlugin(nx)
    registry.register(plugin)

    # Test plugin is registered
    assert registry.get_plugin("my-plugin") is not None

    # Test commands
    await plugin.hello_command("Test")

    # Cleanup
    nx.close()

# Enable debug logging
import logging
logging.basicConfig(level=logging.DEBUG)

from nexus.plugins import PluginRegistry
registry = PluginRegistry()
discovered = registry.discover()  # Check logs for errors