Enforce plugin contract purity: single import surface via shopdb.api

Plugins were reaching into internal core paths (shopdb.core.models.*,
shopdb.extensions, shopdb.utils.*), coupling them to core's file layout and
violating the ADR-001 contract. Consolidate onto one versioned surface.

- shopdb.api: expand from 2 helpers to the full plugin import surface -
  db, cache; BaseModel, AuditMixin; core models (Asset, AssetType,
  AssetStatus, Vendor, Model, Communication, CommunicationType, Location,
  Setting, AuditLog, Application, AppVersion, OperatingSystem); response +
  pagination helpers; employee_connection. Documented in PLUGIN-HOOKS.md.
- Migrate all 22 plugin source files to import only from shopdb.api (plus
  shopdb.plugins.base for the ABC).
- Drop the printers plugin's legacy MachineType dependency: remove
  _ensure_legacy_machine_types and the seed_supplies machinetypeid lookup
  (Model.machinetypeid is nullable; printers carry type via PrinterType).
- Guard test test_plugins_only_import_contract_surface scans plugin source
  and fails on any core import outside shopdb.api / shopdb.plugins.base.
- Scaffold templates updated so generated plugins are contract-pure.
- Bump __contract_version__ 0.2.0 -> 0.3.0 (additive surface expansion;
  manifests pin <1.0.0 so they still satisfy).

145 tests pass, naming/style green, app factory boots all 6 plugins.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-06-26 16:45:06 -04:00
parent 37ffb4add5
commit f663cc5bbe
29 changed files with 2152 additions and 2107 deletions

View File

@@ -1,307 +1,304 @@
"""Computers plugin main class."""
import json
import logging
from pathlib import Path
from typing import List, Dict, Optional, Type
from flask import Flask, Blueprint
import click
from shopdb.plugins.base import BasePlugin, PluginMeta
from shopdb.extensions import db
from shopdb.core.models import AssetType, AssetStatus
from .models import Computer, ComputerType, ComputerInstalledApp
from .api import computers_bp
logger = logging.getLogger(__name__)
class ComputersPlugin(BasePlugin):
"""
Computers plugin - manages PC, server, and workstation assets.
Computers include shopfloor PCs, engineer workstations, servers, etc.
Uses the new Asset architecture with Computer extension table.
"""
def __init__(self):
self._manifest = self._load_manifest()
def _load_manifest(self) -> Dict:
"""Load plugin manifest from JSON file."""
manifestpath = Path(__file__).parent / 'manifest.json'
if manifestpath.exists():
with open(manifestpath, 'r') as f:
return json.load(f)
return {}
@property
def meta(self) -> PluginMeta:
"""Return plugin metadata."""
return PluginMeta(
name=self._manifest.get('name', 'computers'),
version=self._manifest.get('version', '1.0.0'),
description=self._manifest.get(
'description',
'Computer management for PCs, servers, and workstations'
),
author=self._manifest.get('author', 'ShopDB Team'),
dependencies=self._manifest.get('dependencies', []),
core_version=self._manifest.get('core_version', '>=1.0.0'),
api_prefix=self._manifest.get('api_prefix', '/api/computers'),
)
def get_blueprint(self) -> Optional[Blueprint]:
"""Return Flask Blueprint with API routes."""
return computers_bp
def get_models(self) -> List[Type]:
"""Return list of SQLAlchemy model classes."""
return [Computer, ComputerType, ComputerInstalledApp]
def init_app(self, app: Flask, db_instance) -> None:
"""Initialize plugin with Flask app."""
logger.info(f"Computers plugin initialized (v{self.meta.version})")
# -- ADR-006 collector contract -----------------------------------------
def get_collector_schema(self) -> Optional[Dict]:
"""Schema for the PC collector payload (matched by hostname)."""
return {
'identityfield': 'hostname',
'fields': {
'hostname': {'type': 'string', 'required': True},
'serialnumber': {'type': 'string'},
'currentuser': {'type': 'string'},
'lastboottime': {'type': 'string', 'format': 'date-time'},
'ipaddress': {'type': 'string'},
'installedsoftware': {
'type': 'array',
'items': {'name': 'string', 'version': 'string'},
},
},
}
def apply_collector_payload(self, payload: Dict) -> Dict:
"""Idempotent upsert of a PC from a collector payload (by hostname)."""
from datetime import datetime
from shopdb.core.models import (
Asset, AssetType, Application, Communication, CommunicationType,
)
warnings = []
hostname = (payload.get('hostname') or '').strip()
if not hostname:
raise ValueError('hostname is required')
comp = Computer.query.filter(Computer.hostname.ilike(hostname)).first()
if not comp:
comp = (Computer.query.join(Asset, Asset.assetid == Computer.assetid)
.filter(Asset.assetnumber.ilike(hostname)).first())
action = 'updated'
if not comp:
atype = AssetType.query.filter_by(assettype='computer').first()
asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
statusid=1)
db.session.add(asset)
db.session.flush()
comp = Computer(assetid=asset.assetid, hostname=hostname)
db.session.add(comp)
db.session.flush()
action = 'created'
comp.lastreporteddate = datetime.utcnow()
if payload.get('lastboottime'):
try:
comp.lastboottime = datetime.fromisoformat(
payload['lastboottime'].replace('Z', '+00:00'))
except (ValueError, AttributeError):
warnings.append('lastboottime not parseable')
if payload.get('currentuser'):
comp.loggedinuser = payload['currentuser']
if payload.get('serialnumber') and comp.asset:
comp.asset.serialnumber = payload['serialnumber']
if payload.get('ipaddress'):
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
primary = Communication.query.filter_by(
assetid=comp.assetid, isprimary=True).first()
if primary:
primary.ipaddress = payload['ipaddress']
elif ip_comtype:
db.session.add(Communication(
assetid=comp.assetid, comtypeid=ip_comtype.comtypeid,
ipaddress=payload['ipaddress'], isprimary=True))
for app_data in payload.get('installedsoftware', []) or []:
name = app_data.get('name')
if not name:
continue
app = Application.query.filter(Application.appname.ilike(name)).first()
if not app:
warnings.append(f'unknown application: {name}')
continue
installed = ComputerInstalledApp.query.filter_by(
computerid=comp.computerid, appid=app.appid).first()
version = app_data.get('version')
if installed:
installed.installedversion = version
installed.isactive = True
else:
db.session.add(ComputerInstalledApp(
computerid=comp.computerid, appid=app.appid,
installedversion=version))
db.session.commit()
return {
'action': action,
'assetid': comp.assetid,
'identityvalue': hostname,
'warnings': warnings,
}
def on_install(self, app: Flask) -> None:
"""Called when plugin is installed."""
with app.app_context():
self._ensure_asset_type()
self._ensure_computer_types()
logger.info("Computers plugin installed")
def _ensure_asset_type(self) -> None:
"""Ensure computer asset type exists."""
existing = AssetType.query.filter_by(assettype='computer').first()
if not existing:
at = AssetType(
assettype='computer',
pluginname='computers',
tablename='computers',
description='PCs, servers, and workstations',
icon='desktop'
)
db.session.add(at)
logger.debug("Created asset type: computer")
db.session.commit()
def _ensure_computer_types(self) -> None:
"""Ensure basic computer types exist."""
computer_types = [
('Shopfloor PC', 'PC located on the shop floor for machine operation', 'desktop'),
('Engineer Workstation', 'Engineering workstation for CAD/CAM work', 'laptop'),
('CMM PC', 'PC dedicated to CMM operation', 'desktop'),
('Server', 'Server system', 'server'),
('Kiosk', 'Kiosk or info display PC', 'tv'),
('Laptop', 'Laptop computer', 'laptop'),
('Virtual Machine', 'Virtual machine', 'cloud'),
('Other', 'Other computer type', 'desktop'),
]
for name, description, icon in computer_types:
existing = ComputerType.query.filter_by(computertype=name).first()
if not existing:
ct = ComputerType(
computertype=name,
description=description,
icon=icon
)
db.session.add(ct)
logger.debug(f"Created computer type: {name}")
db.session.commit()
def on_uninstall(self, app: Flask) -> None:
"""Called when plugin is uninstalled."""
logger.info("Computers plugin uninstalled")
def get_cli_commands(self) -> List:
"""Return CLI commands for this plugin."""
@click.group('computers')
def computerscli():
"""Computers plugin commands."""
pass
@computerscli.command('list-types')
def list_types():
"""List all computer types."""
from flask import current_app
with current_app.app_context():
types = ComputerType.query.filter_by(isactive=True).all()
if not types:
click.echo('No computer types found.')
return
click.echo('Computer Types:')
for t in types:
click.echo(f" [{t.computertypeid}] {t.computertype}")
@computerscli.command('stats')
def stats():
"""Show computer statistics."""
from flask import current_app
from shopdb.core.models import Asset
with current_app.app_context():
total = db.session.query(Computer).join(Asset).filter(
Asset.isactive == True
).count()
click.echo(f"Total active computers: {total}")
# Shopfloor count
shopfloor = db.session.query(Computer).join(Asset).filter(
Asset.isactive == True,
Computer.isshopfloor == True
).count()
click.echo(f" Shopfloor PCs: {shopfloor}")
click.echo(f" Other: {total - shopfloor}")
@computerscli.command('find')
@click.argument('hostname')
def find_by_hostname(hostname):
"""Find a computer by hostname."""
from flask import current_app
with current_app.app_context():
comp = Computer.query.filter(
Computer.hostname.ilike(f'%{hostname}%')
).first()
if not comp:
click.echo(f'No computer found matching hostname: {hostname}')
return
click.echo(f'Found: {comp.hostname}')
click.echo(f' Asset: {comp.asset.assetnumber}')
click.echo(f' Type: {comp.computertype.computertype if comp.computertype else "N/A"}')
click.echo(f' OS: {comp.operatingsystem.osname if comp.operatingsystem else "N/A"}')
click.echo(f' Logged in: {comp.loggedinuser or "N/A"}')
return [computerscli]
def get_dashboard_widgets(self) -> List[Dict]:
"""Return dashboard widget definitions."""
return [
{
'name': 'Computer Status',
'component': 'ComputerStatusWidget',
'endpoint': '/api/computers/dashboard/summary',
'size': 'medium',
'position': 6,
},
]
def get_navigation_items(self) -> List[Dict]:
"""Return navigation menu items."""
return [
{
'name': 'PCs',
'icon': 'desktop',
'route': '/pcs',
'position': 15,
},
]
"""Computers plugin main class."""
import json
import logging
from pathlib import Path
from typing import List, Dict, Optional, Type
from flask import Flask, Blueprint
import click
from shopdb.plugins.base import BasePlugin, PluginMeta
from shopdb.api import db, AssetType, AssetStatus
from .models import Computer, ComputerType, ComputerInstalledApp
from .api import computers_bp
logger = logging.getLogger(__name__)
class ComputersPlugin(BasePlugin):
"""
Computers plugin - manages PC, server, and workstation assets.
Computers include shopfloor PCs, engineer workstations, servers, etc.
Uses the new Asset architecture with Computer extension table.
"""
def __init__(self):
self._manifest = self._load_manifest()
def _load_manifest(self) -> Dict:
"""Load plugin manifest from JSON file."""
manifestpath = Path(__file__).parent / 'manifest.json'
if manifestpath.exists():
with open(manifestpath, 'r') as f:
return json.load(f)
return {}
@property
def meta(self) -> PluginMeta:
"""Return plugin metadata."""
return PluginMeta(
name=self._manifest.get('name', 'computers'),
version=self._manifest.get('version', '1.0.0'),
description=self._manifest.get(
'description',
'Computer management for PCs, servers, and workstations'
),
author=self._manifest.get('author', 'ShopDB Team'),
dependencies=self._manifest.get('dependencies', []),
core_version=self._manifest.get('core_version', '>=1.0.0'),
api_prefix=self._manifest.get('api_prefix', '/api/computers'),
)
def get_blueprint(self) -> Optional[Blueprint]:
"""Return Flask Blueprint with API routes."""
return computers_bp
def get_models(self) -> List[Type]:
"""Return list of SQLAlchemy model classes."""
return [Computer, ComputerType, ComputerInstalledApp]
def init_app(self, app: Flask, db_instance) -> None:
"""Initialize plugin with Flask app."""
logger.info(f"Computers plugin initialized (v{self.meta.version})")
# -- ADR-006 collector contract -----------------------------------------
def get_collector_schema(self) -> Optional[Dict]:
"""Schema for the PC collector payload (matched by hostname)."""
return {
'identityfield': 'hostname',
'fields': {
'hostname': {'type': 'string', 'required': True},
'serialnumber': {'type': 'string'},
'currentuser': {'type': 'string'},
'lastboottime': {'type': 'string', 'format': 'date-time'},
'ipaddress': {'type': 'string'},
'installedsoftware': {
'type': 'array',
'items': {'name': 'string', 'version': 'string'},
},
},
}
def apply_collector_payload(self, payload: Dict) -> Dict:
"""Idempotent upsert of a PC from a collector payload (by hostname)."""
from datetime import datetime
from shopdb.api import Asset, AssetType, Application, Communication, CommunicationType
warnings = []
hostname = (payload.get('hostname') or '').strip()
if not hostname:
raise ValueError('hostname is required')
comp = Computer.query.filter(Computer.hostname.ilike(hostname)).first()
if not comp:
comp = (Computer.query.join(Asset, Asset.assetid == Computer.assetid)
.filter(Asset.assetnumber.ilike(hostname)).first())
action = 'updated'
if not comp:
atype = AssetType.query.filter_by(assettype='computer').first()
asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
statusid=1)
db.session.add(asset)
db.session.flush()
comp = Computer(assetid=asset.assetid, hostname=hostname)
db.session.add(comp)
db.session.flush()
action = 'created'
comp.lastreporteddate = datetime.utcnow()
if payload.get('lastboottime'):
try:
comp.lastboottime = datetime.fromisoformat(
payload['lastboottime'].replace('Z', '+00:00'))
except (ValueError, AttributeError):
warnings.append('lastboottime not parseable')
if payload.get('currentuser'):
comp.loggedinuser = payload['currentuser']
if payload.get('serialnumber') and comp.asset:
comp.asset.serialnumber = payload['serialnumber']
if payload.get('ipaddress'):
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
primary = Communication.query.filter_by(
assetid=comp.assetid, isprimary=True).first()
if primary:
primary.ipaddress = payload['ipaddress']
elif ip_comtype:
db.session.add(Communication(
assetid=comp.assetid, comtypeid=ip_comtype.comtypeid,
ipaddress=payload['ipaddress'], isprimary=True))
for app_data in payload.get('installedsoftware', []) or []:
name = app_data.get('name')
if not name:
continue
app = Application.query.filter(Application.appname.ilike(name)).first()
if not app:
warnings.append(f'unknown application: {name}')
continue
installed = ComputerInstalledApp.query.filter_by(
computerid=comp.computerid, appid=app.appid).first()
version = app_data.get('version')
if installed:
installed.installedversion = version
installed.isactive = True
else:
db.session.add(ComputerInstalledApp(
computerid=comp.computerid, appid=app.appid,
installedversion=version))
db.session.commit()
return {
'action': action,
'assetid': comp.assetid,
'identityvalue': hostname,
'warnings': warnings,
}
def on_install(self, app: Flask) -> None:
"""Called when plugin is installed."""
with app.app_context():
self._ensure_asset_type()
self._ensure_computer_types()
logger.info("Computers plugin installed")
def _ensure_asset_type(self) -> None:
"""Ensure computer asset type exists."""
existing = AssetType.query.filter_by(assettype='computer').first()
if not existing:
at = AssetType(
assettype='computer',
pluginname='computers',
tablename='computers',
description='PCs, servers, and workstations',
icon='desktop'
)
db.session.add(at)
logger.debug("Created asset type: computer")
db.session.commit()
def _ensure_computer_types(self) -> None:
"""Ensure basic computer types exist."""
computer_types = [
('Shopfloor PC', 'PC located on the shop floor for machine operation', 'desktop'),
('Engineer Workstation', 'Engineering workstation for CAD/CAM work', 'laptop'),
('CMM PC', 'PC dedicated to CMM operation', 'desktop'),
('Server', 'Server system', 'server'),
('Kiosk', 'Kiosk or info display PC', 'tv'),
('Laptop', 'Laptop computer', 'laptop'),
('Virtual Machine', 'Virtual machine', 'cloud'),
('Other', 'Other computer type', 'desktop'),
]
for name, description, icon in computer_types:
existing = ComputerType.query.filter_by(computertype=name).first()
if not existing:
ct = ComputerType(
computertype=name,
description=description,
icon=icon
)
db.session.add(ct)
logger.debug(f"Created computer type: {name}")
db.session.commit()
def on_uninstall(self, app: Flask) -> None:
"""Called when plugin is uninstalled."""
logger.info("Computers plugin uninstalled")
def get_cli_commands(self) -> List:
"""Return CLI commands for this plugin."""
@click.group('computers')
def computerscli():
"""Computers plugin commands."""
pass
@computerscli.command('list-types')
def list_types():
"""List all computer types."""
from flask import current_app
with current_app.app_context():
types = ComputerType.query.filter_by(isactive=True).all()
if not types:
click.echo('No computer types found.')
return
click.echo('Computer Types:')
for t in types:
click.echo(f" [{t.computertypeid}] {t.computertype}")
@computerscli.command('stats')
def stats():
"""Show computer statistics."""
from flask import current_app
from shopdb.api import Asset
with current_app.app_context():
total = db.session.query(Computer).join(Asset).filter(
Asset.isactive == True
).count()
click.echo(f"Total active computers: {total}")
# Shopfloor count
shopfloor = db.session.query(Computer).join(Asset).filter(
Asset.isactive == True,
Computer.isshopfloor == True
).count()
click.echo(f" Shopfloor PCs: {shopfloor}")
click.echo(f" Other: {total - shopfloor}")
@computerscli.command('find')
@click.argument('hostname')
def find_by_hostname(hostname):
"""Find a computer by hostname."""
from flask import current_app
with current_app.app_context():
comp = Computer.query.filter(
Computer.hostname.ilike(f'%{hostname}%')
).first()
if not comp:
click.echo(f'No computer found matching hostname: {hostname}')
return
click.echo(f'Found: {comp.hostname}')
click.echo(f' Asset: {comp.asset.assetnumber}')
click.echo(f' Type: {comp.computertype.computertype if comp.computertype else "N/A"}')
click.echo(f' OS: {comp.operatingsystem.osname if comp.operatingsystem else "N/A"}')
click.echo(f' Logged in: {comp.loggedinuser or "N/A"}')
return [computerscli]
def get_dashboard_widgets(self) -> List[Dict]:
"""Return dashboard widget definitions."""
return [
{
'name': 'Computer Status',
'component': 'ComputerStatusWidget',
'endpoint': '/api/computers/dashboard/summary',
'size': 'medium',
'position': 6,
},
]
def get_navigation_items(self) -> List[Dict]:
"""Return navigation menu items."""
return [
{
'name': 'PCs',
'icon': 'desktop',
'route': '/pcs',
'position': 15,
},
]