Files
shopdb-flask/shopdb/plugins/cli.py
cproudlock 8eb9362452 Phase 4: plugin scaffolding (flask plugin new) with canary tests
Lowers the barrier for sister sites to build their own plugins.
Generated output satisfies the framework contract out of the box.

CLI command (shopdb/plugins/cli.py):
- `flask plugin new <name> --description "..."` generates a plugin
  skeleton under plugins/<name>/. Validates the name against
  CONTRIBUTING.md rules (lowercase letters/digits only, no
  underscores or hyphens, not in the reserved list) and refuses to
  overwrite existing plugins unless --overwrite is passed.
- Output prints the next steps (install, migrate, test).

Scaffolder (shopdb/plugins/scaffolder.py):
- validate_name: enforces the naming rules
- pascal_case: lowercase-to-PascalCase for class names
- scaffold_plugin: copies templates with string.Template
  substitution. Three placeholders: $name, $Name, $description.
  Files with `model.py` in the path get renamed to <name>.py.

Templates (shopdb/plugins/templates/):
- manifest.json.tmpl: name, version 0.1.0, description, core_version
  range >=0.1.0,<1.0.0 (broad enough to survive minor framework bumps)
- plugin.py.tmpl: <Name>Plugin class extending BasePlugin with all
  required hooks implemented (meta from manifest, get_blueprint
  returning the bp, get_models returning the example model). Includes
  on_install hook that seeds the AssetType row.
- models/__init__.py.tmpl + models/model.py.tmpl: Asset extension
  table keyed by assetid with one example field. TODO comment marks
  it as a placeholder.
- api/__init__.py.tmpl + api/routes.py.tmpl: Blueprint with list and
  detail endpoints using the framework's pagination + response helpers.
- schemas/__init__.py.tmpl: marshmallow schema stub.
- tests/__init__.py.tmpl + tests/test_plugin.py.tmpl: smoke tests
  asserting plugin loads, get_blueprint returns Blueprint, get_models
  returns at least one model.
- README.md.tmpl: one-pager for plugin authors with common edits and
  next-step references.

Canary tests (tests/test_plugin_scaffold.py):
- 14 tests asserting the scaffold output passes contract checks.
- Validates name rules (lowercase, reserved, hyphens, digits, etc.)
- Verifies all expected files generated, manifest fields present.
- Loads the generated plugin via PluginLoader (spec_from_file_location
  bypasses the real `plugins` package shadowing).
- Asserts subclasses BasePlugin, get_blueprint returns Blueprint,
  get_models returns model with __tablename__.
- Module-scoped fixture; cleans up sys.modules + SQLAlchemy metadata
  on teardown to avoid cross-test contamination.

Quickstart docs (docs/PLUGIN-QUICKSTART.md):
- 30-minute walkthrough: scaffold -> edit model -> add routes ->
  install -> verify -> add hooks. Cross-links to PLUGIN-HOOKS.md and
  the ADRs. Includes common-errors table.

Test count: 87 -> 101 passing.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-08 17:13:46 -04:00

244 lines
7.8 KiB
Python

"""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'))
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. Run: flask plugin install {name}')
click.echo(f' 4. Run: flask db migrate -m "Add {name} plugin"')
click.echo(f' 5. Run: flask db upgrade')
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)