shopdb-flask/shopdb/plugins/cli.py

"""Flask CLI commands for plugin management."""

import click
from flask import current_app
from flask.cli import with_appcontext


@click.group('plugin')
def plugin_cli():
    """Plugin management commands."""
    pass


@plugin_cli.command('list')
@with_appcontext
def list_plugins():
    """List all available plugins."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        return

    plugins = pm.discover_available()

    if not plugins:
        click.echo("No plugins found in plugins directory.")
        return

    # Format output
    click.echo("")
    click.echo(click.style("Available Plugins:", fg='cyan', bold=True))
    click.echo("-" * 60)

    for p in plugins:
        if p['enabled']:
            status = click.style("[Enabled]", fg='green')
        elif p['installed']:
            status = click.style("[Disabled]", fg='yellow')
        else:
            status = click.style("[Available]", fg='white')

        click.echo(f"  {p['name']:20} v{p['version']:10} {status}")
        if p['description']:
            click.echo(f"    {p['description'][:55]}...")
        if p['dependencies']:
            deps = ', '.join(p['dependencies'])
            click.echo(f"    Dependencies: {deps}")

    click.echo("")


@plugin_cli.command('install')
@click.argument('name')
@click.option('--skip-migrations', is_flag=True, help='Skip database migrations')
@with_appcontext
def install_plugin(name: str, skip_migrations: bool):
    """
    Install a plugin.

    Usage: flask plugin install printers
    """
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    click.echo(f"Installing plugin: {name}")

    if pm.install_plugin(name, run_migrations=not skip_migrations):
        click.echo(click.style(f"Successfully installed {name}", fg='green'))
    else:
        click.echo(click.style(f"Failed to install {name}", fg='red'))
        raise SystemExit(1)


@plugin_cli.command('uninstall')
@click.argument('name')
@click.option('--remove-data', is_flag=True, help='Remove plugin database tables')
@click.confirmation_option(prompt='Are you sure you want to uninstall this plugin?')
@with_appcontext
def uninstall_plugin(name: str, remove_data: bool):
    """
    Uninstall a plugin.

    Usage: flask plugin uninstall printers
    """
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    click.echo(f"Uninstalling plugin: {name}")

    if pm.uninstall_plugin(name, remove_data=remove_data):
        click.echo(click.style(f"Successfully uninstalled {name}", fg='green'))
    else:
        click.echo(click.style(f"Failed to uninstall {name}", fg='red'))
        raise SystemExit(1)


@plugin_cli.command('enable')
@click.argument('name')
@with_appcontext
def enable_plugin(name: str):
    """Enable a disabled plugin."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    if pm.enable_plugin(name):
        click.echo(click.style(f"Enabled {name}", fg='green'))
        # Nudge the operator if the plugin's chain is ahead of the DB.
        try:
            if pm.migration_manager and \
                    pm.migration_manager.has_unapplied_migrations(name):
                click.echo(click.style(
                    f"  {name} has unapplied migrations - "
                    f"run 'flask plugin upgrade-all'", fg='yellow'))
        except Exception:
            pass
    else:
        click.echo(click.style(f"Failed to enable {name}", fg='red'))
        raise SystemExit(1)


@plugin_cli.command('disable')
@click.argument('name')
@with_appcontext
def disable_plugin(name: str):
    """Disable an enabled plugin."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    if pm.disable_plugin(name):
        click.echo(click.style(f"Disabled {name}", fg='green'))
    else:
        click.echo(click.style(f"Failed to disable {name}", fg='red'))
        raise SystemExit(1)


@plugin_cli.command('info')
@click.argument('name')
@with_appcontext
def plugin_info(name: str):
    """Show detailed information about a plugin."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    plugin_class = pm.loader.load_plugin_class(name)
    if not plugin_class:
        click.echo(click.style(f"Plugin {name} not found", fg='red'))
        raise SystemExit(1)

    try:
        temp = plugin_class()
        meta = temp.meta
    except Exception as e:
        click.echo(click.style(f"Error loading plugin: {e}", fg='red'))
        raise SystemExit(1)

    state = pm.registry.get(name)

    click.echo("")
    click.echo("=" * 50)
    click.echo(click.style(f"Plugin: {meta.name}", fg='cyan', bold=True))
    click.echo("=" * 50)
    click.echo(f"Version:      {meta.version}")
    click.echo(f"Description:  {meta.description}")
    click.echo(f"Author:       {meta.author or 'Unknown'}")
    click.echo(f"API Prefix:   {meta.api_prefix}")
    click.echo(f"Dependencies: {', '.join(meta.dependencies) or 'None'}")
    click.echo(f"Core Version: {meta.core_version}")
    click.echo("")

    if state:
        status = click.style('Enabled', fg='green') if state.enabled else click.style('Disabled', fg='yellow')
        click.echo(f"Status:       {status}")
        click.echo(f"Installed:    {state.installed_at}")
        click.echo(f"Migrations:   {len(state.migrations_applied)} applied")
    else:
        click.echo(f"Status:       {click.style('Not installed', fg='white')}")

    click.echo("")


@plugin_cli.command('new')
@click.argument('name')
@click.option('--description', default='', help='One-sentence plugin description')
@click.option('--overwrite', is_flag=True, help='Overwrite existing plugin directory')
@with_appcontext
def new_plugin(name: str, description: str, overwrite: bool):
    """Scaffold a new plugin from the bundled templates.

    Usage: flask plugin new cameras --description "Tracks shop-floor cameras"
    """
    from pathlib import Path
    from .scaffolder import scaffold_plugin, ScaffoldError

    plugins_dir = Path(current_app.root_path).parent / 'plugins'

    if not description:
        description = f'{name.capitalize()} plugin (TODO: replace this description)'

    try:
        target = scaffold_plugin(
            name=name,
            description=description,
            plugins_dir=plugins_dir,
            overwrite=overwrite,
        )
    except ScaffoldError as e:
        click.echo(click.style(f'Scaffold failed: {e}', fg='red'))
        raise SystemExit(1)

    click.echo(click.style(f'Created plugin at {target}', fg='green'))
    click.echo('')
    click.echo('Next steps:')
    click.echo(f'  1. Edit plugins/{name}/models/{name}.py with your domain fields')
    click.echo(f'  2. Edit plugins/{name}/api/routes.py with your endpoints')
    click.echo(f'  3. Add plugins/{name}/migrations/ with an Alembic chain that')
    click.echo(f'     creates your tables (per-plugin chain, NOT the core chain;')
    click.echo(f'     see ADR-008). Register the tables in PLUGIN_TABLE_OWNERS.')
    click.echo(f'  4. Run: flask plugin install {name}')
    click.echo(f'  5. Run: flask plugin upgrade-all')
    click.echo(f'  6. Run: pytest plugins/{name}/tests/')


@plugin_cli.command('migrate')
@click.argument('name')
@click.option('--revision', default='head', help='Target revision')
@with_appcontext
def migrate_plugin(name: str, revision: str):
    """Run migrations for a specific plugin."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    if not pm.registry.is_installed(name):
        click.echo(click.style(f"Plugin {name} is not installed", fg='red'))
        raise SystemExit(1)

    click.echo(f"Running migrations for {name}...")

    if pm.migration_manager.run_plugin_migrations(name, revision):
        click.echo(click.style("Migrations completed", fg='green'))
    else:
        click.echo(click.style("Migration failed", fg='red'))
        raise SystemExit(1)


@plugin_cli.command('upgrade-all')
@with_appcontext
def upgrade_all_plugins():
    """Run pending migrations for every discovered plugin.

    Idempotent. Run this after `flask db upgrade` on every deploy and
    upgrade. It stamps each bundled plugin's anchor revision into
    alembic_version_<plugin> and applies any per-plugin migrations added
    after the ownership cutover (ADR-008). Safe to re-run at head.
    """
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)
    results = pm.upgrade_all_plugins()
    if not results:
        click.echo("No plugins discovered.")
        return
    for name, status in sorted(results.items()):
        if status == 'ok':
            click.echo(click.style(f"  {name:20} ok", fg='green'))
        elif status == 'no-migrations':
            click.echo(click.style(f"  {name:20} no migrations", fg='yellow'))
        else:
            click.echo(click.style(f"  {name:20} {status}", fg='red'))