Files
shopdb-flask/shopdb/plugins/base.py
cproudlock 22e623c1f6 Plugin framework maturation, reports overhaul, theming, and USB frontend repair
Framework:
- Per-plugin Alembic migration chains (ADR-008): every bundled plugin
  carries its own chain with a stamp-only anchor at the ownership cutover;
  new plugin schema lands in plugins/<name>/migrations/, never the core
  chain. Deploys add flask plugin upgrade-all. Fixed a latent bug in the
  shared alembic template (engine URL resolution) and taught the metadata
  filter to include FK-referenced core tables.
- Frontend plugin route gating (ADR-009): plugin routes carry meta.plugin;
  a disabled plugin's pages redirect to the dashboard via a cached,
  fail-open check against the new public GET /api/plugins/enabled.
- get_reports() plugin hook (contract 0.5.0 -> 0.6.0): plugins contribute
  report cards; warranty and toner cards moved off the hardcoded list.

Reports:
- Hub grouped by category with search; inline reports render at the top,
  are URL-backed (?report=id, back-button and deep links work), expose
  their server-side filter params as controls, and export CSV. Warranty
  and Toner pages gained CSV export.
- Deleted the dead legacy Warranty Status report (always-zero buckets
  from a retired column).

Theming and fonts:
- Inter (variable) bundled locally via @fontsource, replacing the Google
  Fonts Roboto import - air-gapped installs now render correctly; tables
  use tabular numerals.
- Optional brand_primary_dark_color, brand_accent_color,
  brand_sidebar_color settings applied to CSS vars at bootstrap.

USB frontend repair (views were reading a dead legacy shape):
- List/detail/form and the employee profile USB panels remapped to the
  real API shape (device_id/device_desc/checkinoutlog); employee panels
  now use /usb/checkouts endpoints; external-mode /usb/checkouts/active
  honors the badge filter; dead client methods pruned.

Also: warranties list page no longer requires login (matches app
convention); collector doc rewritten with a GE-Enforce integration guide
and paste-ready PowerShell reporter; ADR index and CHANGELOG updated.

Verified: 323 tests pass, naming/style green, frontend builds, plugin
migration dry-run green on scratch MySQL.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 10:01:47 -04:00

230 lines
7.8 KiB
Python

"""Base plugin class that all plugins must inherit from."""
from abc import ABC, abstractmethod
from typing import List, Dict, Optional, Type
from dataclasses import dataclass, field
from flask import Flask, Blueprint
@dataclass
class PluginMeta:
"""Plugin metadata container."""
name: str
version: str
description: str
author: str = ""
dependencies: List[str] = field(default_factory=list)
# Default to the current pre-1.0 contract range; tighten when 1.0 lands
core_version: str = ">=0.2.0,<1.0.0"
api_prefix: str = None
def __post_init__(self):
if self.api_prefix is None:
self.api_prefix = f"/api/{self.name.replace('_', '-')}"
class BasePlugin(ABC):
"""
Base class for all ShopDB plugins.
Plugins must implement:
- meta: PluginMeta instance
- get_blueprint(): Return Flask Blueprint for API routes
- get_models(): Return list of SQLAlchemy model classes
Optionally implement:
- init_app(app, db): Custom initialization
- get_cli_commands(): Return Click commands
- get_services(): Return service classes
- on_install(): Called when plugin is installed
- on_uninstall(): Called when plugin is uninstalled
- on_enable(): Called when plugin is enabled
- on_disable(): Called when plugin is disabled
"""
@property
@abstractmethod
def meta(self) -> PluginMeta:
"""Return plugin metadata."""
pass
@abstractmethod
def get_blueprint(self) -> Optional[Blueprint]:
"""Return Flask Blueprint with API routes."""
pass
@abstractmethod
def get_models(self) -> List[Type]:
"""Return list of SQLAlchemy model classes."""
pass
def init_app(self, app: Flask, db) -> None:
"""
Initialize plugin with Flask app.
Override for custom initialization.
"""
pass
def get_cli_commands(self) -> List:
"""Return list of Click command groups/commands."""
return []
def get_services(self) -> Dict[str, Type]:
"""Return dict of service name -> service class."""
return {}
def get_provisioning_note(self) -> Optional[Dict]:
"""Transparency note shown when a site enables this plugin.
Return None for plugins that need no special setup. For plugins that
create extra tables (e.g. a self-hosted directory or USB tables), return:
{
'tables': ['directoryemployees', ...], # created in the shopdb DB
'note': 'Plain-language what/why.',
'docs': 'plugins/<name>/README.md', # where the schema lives
}
The setup wizard shows this the moment the plugin is checked.
"""
return None
def get_config_schema(self) -> List[Dict]:
"""Declare the config fields this plugin needs, for the setup wizard.
Each field is a dict:
key - the Setting key (non-secret) it maps to
label - human label
type - 'text' | 'number' | 'password'
secret - True for credentials; these are NOT stored in the DB, the
wizard emits an .env line for the operator to paste instead
envvar - (secret only) the .env variable name to emit
default - optional default shown as a placeholder
help - optional hint
Return [] (default) if the plugin needs no configuration.
"""
return []
def get_setting(self, key: str, default=None):
"""Read a plugin-scoped setting from the core Setting store.
Settings are namespaced by plugin name to avoid collisions
across plugins. The on-disk key is `plugin.<pluginname>.<key>`.
Returns the typed value (string, integer, boolean, etc.) or
the default if not set.
"""
from shopdb.core.models import Setting
namespaced_key = f'plugin.{self.meta.name}.{key}'
return Setting.get(namespaced_key, default)
def set_setting(self, key: str, value, valuetype: str = 'string',
description: str = None) -> None:
"""Write a plugin-scoped setting to the core Setting store.
Settings are namespaced by plugin name. Persists immediately and
survives restarts.
"""
from shopdb.core.models import Setting
namespaced_key = f'plugin.{self.meta.name}.{key}'
Setting.set(
namespaced_key,
value,
valuetype=valuetype,
category=f'plugin.{self.meta.name}',
description=description,
)
def get_collector_schema(self) -> Optional[Dict]:
"""Return JSON Schema describing the collector payload for this plugin.
Return None if the plugin does not accept collector input.
See ADR-006 for the contract. The schema must include:
- 'identityfield': name of the field that uniquely identifies an
asset across submissions (e.g., 'hostname' for PCs,
'macaddress' for network devices). Used for idempotent upsert.
- 'fields': JSON Schema definitions for the rest of the payload.
Plugins returning a non-None schema have an endpoint at
/api/collector/<pluginname> auto-registered by the loader.
"""
return None
def apply_collector_payload(self, payload: Dict) -> Dict:
"""Idempotently upsert an asset from a validated collector payload.
Called by the generic /api/collector/<pluginname> endpoint after the
payload passed schema validation. Plugins that return a schema from
get_collector_schema must implement this. Return a dict with at least:
- 'action': 'created' | 'updated' | 'noop'
- 'assetid': the affected asset id (or None)
- 'warnings': list[str]
"""
raise NotImplementedError(
f"{self.meta.name} declares a collector schema but does not "
f"implement apply_collector_payload"
)
def on_install(self, app: Flask) -> None:
"""Called when plugin is installed via CLI."""
pass
def on_uninstall(self, app: Flask) -> None:
"""Called when plugin is uninstalled via CLI."""
pass
def on_enable(self, app: Flask) -> None:
"""Called when plugin is enabled."""
pass
def on_disable(self, app: Flask) -> None:
"""Called when plugin is disabled."""
pass
def get_dashboard_widgets(self) -> List[Dict]:
"""
Return dashboard widget definitions.
Each widget: {
'name': str,
'component': str, # Frontend component name
'endpoint': str, # API endpoint for data
'size': str, # 'small', 'medium', 'large'
'position': int # Order on dashboard
}
"""
return []
def get_navigation_items(self) -> List[Dict]:
"""
Return navigation menu items.
Each item: {
'name': str,
'icon': str,
'route': str,
'position': int,
'children': []
}
"""
return []
def get_reports(self) -> List[Dict]:
"""
Return report card definitions for the Reports hub.
Each entry: {
'id': str, # stable report id
'name': str, # card title
'description': str, # one-line blurb
'category': str, # grouping key (lowercase)
# plus EXACTLY ONE of:
'route': str, # frontend path for a dedicated report page
'endpoint': str, # API endpoint for generic inline rendering
}
Consumed by GET /api/reports, which merges these after the static core
reports. Disabled plugins are skipped by the consumer.
"""
return []