Files
shopdb-flask/shopdb/plugins/base.py
cproudlock 5fefb53bca Phase 2: define plugin contract surface and add compliance tests
Locks the public surface plugin authors at sister sites depend on.

Contract version (shopdb/__init__.py):
- __contract_version__ = '0.1.0'. Per ADR-002, plugins declare a
  compatible range in manifest.json `core_version`. Pre-1.0 signals
  the contract is still settling; sister sites should pin tightly.

BasePlugin hook changes (shopdb/plugins/base.py):
- Add get_collector_schema() per ADR-006. Returns JSON Schema (with
  identityfield + fields) describing the payload of an external
  collector pushing to /api/collector/<pluginname>. Defaults to None
  (no auto-registered endpoint).
- Remove get_event_handlers(). Event bus deferred indefinitely per
  ADR-001 (no real use case yet; add via new ADR if it appears).

Hook reference (docs/PLUGIN-HOOKS.md):
- Canonical reference for the contract: required hooks (meta,
  get_blueprint, get_models), optional hooks (init_app,
  get_cli_commands, get_services, get_dashboard_widgets,
  get_navigation_items, get_searchable_fields, get_collector_schema),
  lifecycle hooks (on_install, on_uninstall, on_enable, on_disable),
  helpers exposed in shopdb.api (audit_log, Setting,
  resolve_asset_position).
- Versioning rules + change-classification guidance.

Compliance tests (tests/test_plugin_contract.py):
- 8 distinct contract assertions parametrized over 6 bundled plugins
  (computers, equipment, network, notifications, printers, usb).
- Asserts: subclasses BasePlugin; manifest has required fields; meta
  returns valid PluginMeta; get_blueprint returns Blueprint or None;
  get_models returns model classes; get_collector_schema returns
  None or {identityfield, fields}; get_navigation_items and
  get_searchable_fields return list.
- Plus 3 framework-level: __contract_version__ is valid semver,
  get_event_handlers absent, get_collector_schema present.

Test count: 15 -> 66 passing.

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

163 lines
5.0 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)
core_version: str = ">=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_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 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_searchable_fields(self) -> List[Dict]:
"""
Return fields this plugin contributes to global search.
Each field: {
'model': Type, # SQLAlchemy model class
'field': str, # Column name to search
'result_type': str, # Type identifier for search results
'url_template': str, # URL template with {id} placeholder
'title_field': str, # Field to use for result title
'subtitle_field': str, # Optional field for subtitle
'relevance_boost': int # Optional relevance score multiplier
}
Example for equipment plugin:
return [{
'model': Equipment,
'join_model': Asset,
'join_condition': Equipment.assetid == Asset.assetid,
'search_fields': ['assetnumber', 'name', 'serialnumber'],
'result_type': 'equipment',
'url_template': '/equipment/{id}',
'title_field': 'assetnumber',
'subtitle_field': 'name',
}]
"""
return []