Spaces:
Running
Running
File size: 5,613 Bytes
e70c280 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | import os
from pathlib import Path
from typing import Optional
import typer
plugins_app = typer.Typer(name="plugins", help="BrainAPI plugin management")
app = typer.Typer(name="brainapi", help="BrainAPI CLI", no_args_is_help=True)
app.add_typer(plugins_app, name="plugins")
PLUGINS_DIR = Path(os.getenv("PLUGINS_DIR", "plugins"))
REGISTRY_URL = os.getenv("PLUGIN_REGISTRY_URL", "https://registry.brain-api.dev")
PUBLISHER_ID = os.getenv("PLUGIN_PUBLISHER_ID", "")
PUBLISHER_API_KEY = os.getenv("PLUGIN_PUBLISHER_API_KEY", "")
def _get_manager():
from src.core.plugins.manager import PluginManager
return PluginManager(
plugins_dir=PLUGINS_DIR,
registry_url=REGISTRY_URL or None,
publisher_id=PUBLISHER_ID or None,
publisher_api_key=PUBLISHER_API_KEY or None,
)
@plugins_app.command()
def install(
name: str = typer.Argument(..., help="Plugin name to install"),
version: str = typer.Option("latest", "--version", "-v", help="Plugin version"),
):
manager = _get_manager()
try:
manifest = manager.install(name, version=version)
typer.echo(f"Installed {manifest.name} v{manifest.version}")
except Exception as exc:
typer.echo(f"Failed to install '{name}': {exc}", err=True)
raise typer.Exit(code=1)
@plugins_app.command()
def uninstall(
name: str = typer.Argument(..., help="Plugin name to uninstall"),
):
manager = _get_manager()
if manager.uninstall(name):
typer.echo(f"Uninstalled '{name}'")
else:
typer.echo(f"Plugin '{name}' not found", err=True)
raise typer.Exit(code=1)
@plugins_app.command(name="list")
def list_plugins(
remote: bool = typer.Option(False, "--remote", "-r", help="List available plugins from registry"),
):
manager = _get_manager()
if remote:
try:
plugins = manager.list_available()
except Exception as exc:
typer.echo(f"Failed to fetch remote plugins: {exc}", err=True)
raise typer.Exit(code=1)
if not plugins:
typer.echo("No plugins available in the registry")
return
typer.echo("Available plugins:")
for p in plugins:
typer.echo(f" {p.name} v{p.version} - {p.description}")
else:
plugins = manager.list_installed()
if not plugins:
typer.echo("No plugins installed")
return
typer.echo("Installed plugins:")
for p in plugins:
typer.echo(f" {p.name} v{p.version} - {p.description}")
@plugins_app.command()
def info(
name: str = typer.Argument(..., help="Plugin name to get info for"),
):
manager = _get_manager()
try:
meta = manager.get_info(name)
typer.echo(f"Name: {meta.name}")
typer.echo(f"Version: {meta.version}")
typer.echo(f"Author: {meta.author}")
typer.echo(f"Description: {meta.description}")
if meta.versions:
typer.echo(f"Available versions: {', '.join(meta.versions)}")
except Exception as exc:
typer.echo(f"Failed to get info for '{name}': {exc}", err=True)
raise typer.Exit(code=1)
@plugins_app.command()
def update(
name: str = typer.Argument(..., help="Plugin name to update"),
):
manager = _get_manager()
try:
manifest = manager.update(name)
typer.echo(f"Updated {manifest.name} to v{manifest.version}")
except Exception as exc:
typer.echo(f"Failed to update '{name}': {exc}", err=True)
raise typer.Exit(code=1)
@plugins_app.command()
def publish(
archive: Path = typer.Argument(..., help="Path to plugin archive (.tar.gz)"),
name: Optional[str] = typer.Option(None, "--name", "-n", help="Plugin name override"),
force: bool = typer.Option(False, "--force", "-f", help="Overwrite existing version"),
):
manager = _get_manager()
try:
result = manager.publish(archive, name=name, force=force)
typer.echo(f"Published: {result}")
except Exception as exc:
typer.echo(f"Failed to publish: {exc}", err=True)
raise typer.Exit(code=1)
@plugins_app.command()
def depublish(
name: str = typer.Argument(..., help="Plugin name to remove from registry"),
version: Optional[str] = typer.Option(None, "--version", "-v", help="Specific version to remove (omit to remove all)"),
):
manager = _get_manager()
try:
result = manager.delete_remote(name, version=version)
label = f"{name} v{version}" if version else name
typer.echo(f"Removed '{label}' from registry: {result}")
except Exception as exc:
typer.echo(f"Failed to depublish '{name}': {exc}", err=True)
raise typer.Exit(code=1)
@plugins_app.command()
def register(
publisher_id: Optional[str] = typer.Option(None, "--id", help="Desired publisher ID (auto-generated if omitted)"),
):
manager = _get_manager()
try:
result = manager.register_publisher(publisher_id=publisher_id)
typer.echo("")
typer.echo(f" Publisher registered successfully!")
typer.echo(f" Publisher ID: {result['publisher_id']}")
typer.echo(f" API Key: {result['api_key']}")
typer.echo("")
typer.echo(" Save these to your environment:")
typer.echo(f" export PLUGIN_PUBLISHER_ID={result['publisher_id']}")
typer.echo(f" export PLUGIN_PUBLISHER_API_KEY={result['api_key']}")
typer.echo("")
except Exception as exc:
typer.echo(f"Failed to register: {exc}", err=True)
raise typer.Exit(code=1)
if __name__ == "__main__":
app()
|