Files
shopdb-flask/plugins/printedparts/plugin.py
cproudlock df918ed38f
Some checks failed
CI / backend (push) Successful in 1m43s
CI / naming (push) Successful in 1s
CI / frontend (push) Successful in 8s
CI / migrations-mysql (push) Failing after 8s
printedparts stage 11: low-stock email alerts on threshold crossing
Contract 0.12.0: send_email/send_alert join the plugin surface (the
mailer was core-only), PLUGIN-HOOKS and status docs updated, manifest
pins the new floor. The alert fires inside _ledger_write only when a
decrement CROSSES the item's threshold - one alert per depletion,
rearmed by restocking above - and is best-effort after the commit so
mail trouble can never fail a take. Recipients come from
printedparts_alert_email, falling back to the site alert_recipients.
on_enable re-seeds settings idempotently so existing installs pick up
new keys. Crossing/rearm semantics proven by test.
2026-07-17 08:15:50 -04:00

129 lines
4.7 KiB
Python

"""Printedparts plugin main class.
3D-printed parts inventory + kiosk checkout. Quantity-based consumables:
one row is a KIND of part with a count, not an individually tracked asset,
so unlike most plugins this one seeds NO AssetType (ADR-001 assets are
one-row-per-physical-thing). See docs/proposals/printedparts-plugin.md.
"""
import json
import logging
from pathlib import Path
from typing import List, Optional, Type
from flask import Flask, Blueprint
from shopdb.plugins.base import BasePlugin, PluginMeta
from shopdb.api import db, Setting
from .models import PrintedItem, PrintedItemTransaction
from .api import printedparts_bp
logger = logging.getLogger(__name__)
class PrintedpartsPlugin(BasePlugin):
"""3D-printed parts inventory + kiosk checkout."""
def __init__(self):
manifest_path = Path(__file__).parent / 'manifest.json'
with open(manifest_path) as f:
self._manifest = json.load(f)
@property
def meta(self) -> PluginMeta:
return PluginMeta(
name=self._manifest['name'],
version=self._manifest['version'],
description=self._manifest['description'],
author=self._manifest.get('author', ''),
dependencies=self._manifest.get('dependencies', []),
core_version=self._manifest.get('core_version', '>=0.1.0'),
api_prefix=self._manifest.get('api_prefix'),
)
def get_blueprint(self) -> Optional[Blueprint]:
return printedparts_bp
def get_models(self) -> List[Type]:
return [PrintedItem, PrintedItemTransaction]
def init_app(self, app: Flask, db_instance) -> None:
logger.info(f'Printedparts plugin initialized (v{self.meta.version})')
def get_permissions(self) -> List:
"""RBAC permissions this plugin owns (seeded on install/enable)."""
return [
('printedparts.view', 'View 3D printed parts', 'printedparts'),
('printedparts.create', 'Create printed parts', 'printedparts'),
('printedparts.edit', 'Edit printed parts', 'printedparts'),
('printedparts.delete', 'Retire printed parts', 'printedparts'),
('printedparts.restock', 'Restock and adjust stock counts',
'printedparts'),
]
def get_reports(self) -> List[dict]:
return [
{
'id': 'printedparts-stock',
'name': '3D Parts Stock',
'description': 'Stock levels with low-stock flags and the '
'cache-vs-ledger reconcile check',
'category': 'inventory',
'endpoint': '/api/printedparts/reports/stock',
},
{
'id': 'printedparts-consumption',
'name': '3D Parts Consumption',
'description': 'Takes per item over a date range',
'category': 'usage',
'endpoint': '/api/printedparts/reports/consumption',
},
{
'id': 'printedparts-by-person',
'name': '3D Parts by Person',
'description': 'Takes grouped by employee',
'category': 'usage',
'endpoint': '/api/printedparts/reports/by-person',
},
]
def get_navigation_items(self) -> List[dict]:
return [
{
'name': '3D Parts',
'icon': 'box',
'route': '/printedparts',
'position': 46,
},
]
def on_install(self, app: Flask) -> None:
with app.app_context():
self._seed_settings()
logger.info('Printedparts plugin installed')
def on_enable(self, app: Flask) -> None:
# Idempotent re-seed so settings added in later versions reach sites
# that installed earlier (enable runs on every upgrade cycle).
with app.app_context():
self._seed_settings()
def _seed_settings(self) -> None:
defaults = [
('printedparts_code_prefix', '3DP', 'string',
'Prefix for generated item codes'),
('printedparts_default_threshold', '5', 'integer',
'Default low-stock threshold for new items'),
('printedparts_unknown_badge', 'deny', 'string',
'Kiosk policy when a badge resolves to no employee: allow or deny'),
('printedparts_alert_email', '', 'string',
'Comma-separated low-stock alert recipients; empty uses the '
'site alert_recipients'),
]
for key, value, valuetype, description in defaults:
if Setting.get(key) is None:
Setting.set(key, value, valuetype=valuetype,
category='printedparts', description=description)
db.session.commit()