diff --git a/docs/PLUGIN-HOOKS.md b/docs/PLUGIN-HOOKS.md index 87e5606..e38490a 100644 --- a/docs/PLUGIN-HOOKS.md +++ b/docs/PLUGIN-HOOKS.md @@ -233,6 +233,34 @@ These run when the plugin's installation state changes. All optional. | `on_enable(app)` | When the plugin is enabled at runtime | Subscribe to events, warm caches | | `on_disable(app)` | When the plugin is disabled at runtime | Unsubscribe, drain queues | +## The import surface (`shopdb.api`) + +`shopdb.api` is the ONLY core module a plugin may import from (besides +`shopdb.plugins.base` for `BasePlugin` / `PluginMeta`). Importing internal +paths like `shopdb.core.models.*`, `shopdb.extensions`, or `shopdb.utils.*` +is a contract violation and fails the test +`tests/test_plugin_contract.py::test_plugins_only_import_contract_surface`. + +What `shopdb.api` exposes: + +- Infrastructure: `db`, `cache` +- Model bases: `BaseModel`, `AuditMixin` +- Core models: `Asset`, `AssetType`, `AssetStatus`, `Vendor`, `Model`, + `Communication`, `CommunicationType`, `Location`, `Setting`, `AuditLog`, + `Application`, `AppVersion`, `OperatingSystem` +- Responses: `success_response`, `error_response`, `paginated_response`, + `ErrorCodes` +- Pagination: `get_pagination_params`, `paginate_query` +- Helpers: `audit_log`, `resolve_asset_position` +- Legacy employee directory: `employee_connection` + +```python +from shopdb.api import db, Asset, AssetType, success_response, paginate_query +``` + +Adding a name to `shopdb.api` is an additive (minor) contract bump; removing +one is breaking (major). See ADR-002. + ## Helpers exposed to plugins The framework provides helper APIs in `shopdb.api` (the public namespace). diff --git a/plugins/computers/api/routes.py b/plugins/computers/api/routes.py index 9f4f6c5..787cd7e 100644 --- a/plugins/computers/api/routes.py +++ b/plugins/computers/api/routes.py @@ -3,18 +3,7 @@ from flask import Blueprint, request from flask_jwt_extended import jwt_required -from shopdb.extensions import db -from shopdb.core.models import ( - Asset, AssetType, OperatingSystem, Application, AppVersion, AuditLog, - Communication, CommunicationType, -) -from shopdb.utils.responses import ( - success_response, - error_response, - paginated_response, - ErrorCodes -) -from shopdb.utils.pagination import get_pagination_params, paginate_query +from shopdb.api import db, Asset, AssetType, OperatingSystem, Application, AppVersion, AuditLog, Communication, CommunicationType, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query from ..models import Computer, ComputerType, ComputerInstalledApp diff --git a/plugins/computers/models/computer.py b/plugins/computers/models/computer.py index 27325b1..7c2e6f1 100644 --- a/plugins/computers/models/computer.py +++ b/plugins/computers/models/computer.py @@ -1,204 +1,203 @@ -"""Computer plugin models.""" - -from shopdb.extensions import db -from shopdb.core.models.base import BaseModel - - -class ComputerType(BaseModel): - """ - Computer type classification. - - Examples: Shopfloor PC, Engineer Workstation, CMM PC, Server, etc. - """ - __tablename__ = 'computertypes' - - computertypeid = db.Column(db.Integer, primary_key=True) - computertype = db.Column(db.String(100), unique=True, nullable=False) - description = db.Column(db.Text) - icon = db.Column(db.String(50), comment='Icon name for UI') - - def __repr__(self): - return f"" - - -class Computer(BaseModel): - """ - Computer-specific extension data. - - Links to core Asset table via assetid. - Stores computer-specific fields like hostname, OS, logged in user, etc. - """ - __tablename__ = 'computers' - - computerid = db.Column(db.Integer, primary_key=True) - - # Link to core asset - assetid = db.Column( - db.Integer, - db.ForeignKey('assets.assetid', ondelete='CASCADE'), - unique=True, - nullable=False, - index=True - ) - - # Computer classification - computertypeid = db.Column( - db.Integer, - db.ForeignKey('computertypes.computertypeid'), - nullable=True - ) - - # Network identity - hostname = db.Column( - db.String(100), - index=True, - comment='Network hostname' - ) - - # Operating system - osid = db.Column( - db.Integer, - db.ForeignKey('operatingsystems.osid'), - nullable=True - ) - - # Hardware make/model (PCs carry vendor + model like equipment) - vendorid = db.Column( - db.Integer, - db.ForeignKey('vendors.vendorid'), - nullable=True - ) - modelnumberid = db.Column( - db.Integer, - db.ForeignKey('models.modelnumberid'), - nullable=True - ) - - # Status tracking - loggedinuser = db.Column(db.String(100), nullable=True) - lastreporteddate = db.Column(db.DateTime, nullable=True) - lastboottime = db.Column(db.DateTime, nullable=True) - - # Remote access features - isvnc = db.Column( - db.Boolean, - default=False, - comment='VNC remote access enabled' - ) - iswinrm = db.Column( - db.Boolean, - default=False, - comment='WinRM enabled' - ) - - # Classification flags - isshopfloor = db.Column( - db.Boolean, - default=False, - comment='Shopfloor PC (vs office PC)' - ) - - # Relationships - asset = db.relationship( - 'Asset', - backref=db.backref('computer', uselist=False, lazy='joined') - ) - computertype = db.relationship('ComputerType', backref='computers') - operatingsystem = db.relationship('OperatingSystem', backref='computers') - vendor = db.relationship('Vendor') - model = db.relationship('Model') - - # Installed applications (one-to-many) - installedapps = db.relationship( - 'ComputerInstalledApp', - back_populates='computer', - cascade='all, delete-orphan', - lazy='dynamic' - ) - - __table_args__ = ( - db.Index('idx_computer_type', 'computertypeid'), - db.Index('idx_computer_hostname', 'hostname'), - db.Index('idx_computer_os', 'osid'), - ) - - def __repr__(self): - return f"" - - def to_dict(self): - """Convert to dictionary with related names.""" - result = super().to_dict() - - # Add related object names - if self.computertype: - result['computertypename'] = self.computertype.computertype - if self.operatingsystem: - result['osname'] = self.operatingsystem.osname - if self.vendor: - result['vendorname'] = self.vendor.vendor - if self.model: - result['modelname'] = self.model.modelnumber - - return result - - -class ComputerInstalledApp(db.Model): - """ - Junction table for applications installed on computers. - - Tracks which applications are installed on which computers, - including version information. - """ - __tablename__ = 'computerinstalledapps' - - id = db.Column(db.Integer, primary_key=True) - computerid = db.Column( - db.Integer, - db.ForeignKey('computers.computerid', ondelete='CASCADE'), - nullable=False - ) - appid = db.Column( - db.Integer, - db.ForeignKey('applications.appid'), - nullable=False - ) - appversionid = db.Column( - db.Integer, - db.ForeignKey('appversions.appversionid'), - nullable=True - ) - # Raw version string from automated collection (when no curated AppVersion) - installedversion = db.Column(db.String(100), nullable=True) - isactive = db.Column(db.Boolean, default=True, nullable=False) - installeddate = db.Column(db.DateTime, default=db.func.now()) - - # Relationships - computer = db.relationship('Computer', back_populates='installedapps') - application = db.relationship('Application') - appversion = db.relationship('AppVersion') - - __table_args__ = ( - db.UniqueConstraint('computerid', 'appid', name='uq_computer_app'), - db.Index('idx_compapp_computer', 'computerid'), - db.Index('idx_compapp_app', 'appid'), - ) - - def to_dict(self): - """Convert to dictionary.""" - return { - 'id': self.id, - 'computerid': self.computerid, - 'appid': self.appid, - 'appversionid': self.appversionid, - 'isactive': self.isactive, - 'installeddate': self.installeddate.isoformat() + 'Z' if self.installeddate else None, - 'application': { - 'appid': self.application.appid, - 'appname': self.application.appname, - 'appdescription': self.application.appdescription, - } if self.application else None, - 'version': self.appversion.version if self.appversion else None - } - - def __repr__(self): - return f"" +"""Computer plugin models.""" + +from shopdb.api import db, BaseModel + + +class ComputerType(BaseModel): + """ + Computer type classification. + + Examples: Shopfloor PC, Engineer Workstation, CMM PC, Server, etc. + """ + __tablename__ = 'computertypes' + + computertypeid = db.Column(db.Integer, primary_key=True) + computertype = db.Column(db.String(100), unique=True, nullable=False) + description = db.Column(db.Text) + icon = db.Column(db.String(50), comment='Icon name for UI') + + def __repr__(self): + return f"" + + +class Computer(BaseModel): + """ + Computer-specific extension data. + + Links to core Asset table via assetid. + Stores computer-specific fields like hostname, OS, logged in user, etc. + """ + __tablename__ = 'computers' + + computerid = db.Column(db.Integer, primary_key=True) + + # Link to core asset + assetid = db.Column( + db.Integer, + db.ForeignKey('assets.assetid', ondelete='CASCADE'), + unique=True, + nullable=False, + index=True + ) + + # Computer classification + computertypeid = db.Column( + db.Integer, + db.ForeignKey('computertypes.computertypeid'), + nullable=True + ) + + # Network identity + hostname = db.Column( + db.String(100), + index=True, + comment='Network hostname' + ) + + # Operating system + osid = db.Column( + db.Integer, + db.ForeignKey('operatingsystems.osid'), + nullable=True + ) + + # Hardware make/model (PCs carry vendor + model like equipment) + vendorid = db.Column( + db.Integer, + db.ForeignKey('vendors.vendorid'), + nullable=True + ) + modelnumberid = db.Column( + db.Integer, + db.ForeignKey('models.modelnumberid'), + nullable=True + ) + + # Status tracking + loggedinuser = db.Column(db.String(100), nullable=True) + lastreporteddate = db.Column(db.DateTime, nullable=True) + lastboottime = db.Column(db.DateTime, nullable=True) + + # Remote access features + isvnc = db.Column( + db.Boolean, + default=False, + comment='VNC remote access enabled' + ) + iswinrm = db.Column( + db.Boolean, + default=False, + comment='WinRM enabled' + ) + + # Classification flags + isshopfloor = db.Column( + db.Boolean, + default=False, + comment='Shopfloor PC (vs office PC)' + ) + + # Relationships + asset = db.relationship( + 'Asset', + backref=db.backref('computer', uselist=False, lazy='joined') + ) + computertype = db.relationship('ComputerType', backref='computers') + operatingsystem = db.relationship('OperatingSystem', backref='computers') + vendor = db.relationship('Vendor') + model = db.relationship('Model') + + # Installed applications (one-to-many) + installedapps = db.relationship( + 'ComputerInstalledApp', + back_populates='computer', + cascade='all, delete-orphan', + lazy='dynamic' + ) + + __table_args__ = ( + db.Index('idx_computer_type', 'computertypeid'), + db.Index('idx_computer_hostname', 'hostname'), + db.Index('idx_computer_os', 'osid'), + ) + + def __repr__(self): + return f"" + + def to_dict(self): + """Convert to dictionary with related names.""" + result = super().to_dict() + + # Add related object names + if self.computertype: + result['computertypename'] = self.computertype.computertype + if self.operatingsystem: + result['osname'] = self.operatingsystem.osname + if self.vendor: + result['vendorname'] = self.vendor.vendor + if self.model: + result['modelname'] = self.model.modelnumber + + return result + + +class ComputerInstalledApp(db.Model): + """ + Junction table for applications installed on computers. + + Tracks which applications are installed on which computers, + including version information. + """ + __tablename__ = 'computerinstalledapps' + + id = db.Column(db.Integer, primary_key=True) + computerid = db.Column( + db.Integer, + db.ForeignKey('computers.computerid', ondelete='CASCADE'), + nullable=False + ) + appid = db.Column( + db.Integer, + db.ForeignKey('applications.appid'), + nullable=False + ) + appversionid = db.Column( + db.Integer, + db.ForeignKey('appversions.appversionid'), + nullable=True + ) + # Raw version string from automated collection (when no curated AppVersion) + installedversion = db.Column(db.String(100), nullable=True) + isactive = db.Column(db.Boolean, default=True, nullable=False) + installeddate = db.Column(db.DateTime, default=db.func.now()) + + # Relationships + computer = db.relationship('Computer', back_populates='installedapps') + application = db.relationship('Application') + appversion = db.relationship('AppVersion') + + __table_args__ = ( + db.UniqueConstraint('computerid', 'appid', name='uq_computer_app'), + db.Index('idx_compapp_computer', 'computerid'), + db.Index('idx_compapp_app', 'appid'), + ) + + def to_dict(self): + """Convert to dictionary.""" + return { + 'id': self.id, + 'computerid': self.computerid, + 'appid': self.appid, + 'appversionid': self.appversionid, + 'isactive': self.isactive, + 'installeddate': self.installeddate.isoformat() + 'Z' if self.installeddate else None, + 'application': { + 'appid': self.application.appid, + 'appname': self.application.appname, + 'appdescription': self.application.appdescription, + } if self.application else None, + 'version': self.appversion.version if self.appversion else None + } + + def __repr__(self): + return f"" diff --git a/plugins/computers/plugin.py b/plugins/computers/plugin.py index fd42811..6f4e5c0 100644 --- a/plugins/computers/plugin.py +++ b/plugins/computers/plugin.py @@ -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, + }, + ] diff --git a/plugins/equipment/api/routes.py b/plugins/equipment/api/routes.py index ca9ef74..8ca5156 100644 --- a/plugins/equipment/api/routes.py +++ b/plugins/equipment/api/routes.py @@ -3,15 +3,7 @@ from flask import Blueprint, request from flask_jwt_extended import jwt_required -from shopdb.extensions import db -from shopdb.core.models import Asset, AssetType, Vendor, Model, AuditLog -from shopdb.utils.responses import ( - success_response, - error_response, - paginated_response, - ErrorCodes -) -from shopdb.utils.pagination import get_pagination_params, paginate_query +from shopdb.api import db, Asset, AssetType, Vendor, Model, AuditLog, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query from ..models import Equipment, EquipmentType @@ -446,7 +438,7 @@ def dashboard_summary(): ).all() # Count by status - from shopdb.core.models import AssetStatus + from shopdb.api import AssetStatus by_status = db.session.query( AssetStatus.status, db.func.count(Equipment.equipmentid) diff --git a/plugins/equipment/models/equipment.py b/plugins/equipment/models/equipment.py index fe0d2c0..565ed64 100644 --- a/plugins/equipment/models/equipment.py +++ b/plugins/equipment/models/equipment.py @@ -1,133 +1,132 @@ -"""Equipment plugin models.""" - -from shopdb.extensions import db -from shopdb.core.models.base import BaseModel - - -class EquipmentType(BaseModel): - """ - Equipment type classification. - - Examples: CNC, CMM, Lathe, Grinder, EDM, Part Marker, etc. - """ - __tablename__ = 'equipmenttypes' - - equipmenttypeid = db.Column(db.Integer, primary_key=True) - equipmenttype = db.Column(db.String(100), unique=True, nullable=False) - description = db.Column(db.Text) - icon = db.Column(db.String(50), comment='Icon name for UI') - - def __repr__(self): - return f"" - - -class Equipment(BaseModel): - """ - Equipment-specific extension data. - - Links to core Asset table via assetid. - Stores equipment-specific fields like type, model, vendor, etc. - """ - __tablename__ = 'equipment' - - equipmentid = db.Column(db.Integer, primary_key=True) - - # Link to core asset - assetid = db.Column( - db.Integer, - db.ForeignKey('assets.assetid', ondelete='CASCADE'), - unique=True, - nullable=False, - index=True - ) - - # Equipment classification - equipmenttypeid = db.Column( - db.Integer, - db.ForeignKey('equipmenttypes.equipmenttypeid'), - nullable=True - ) - - # Vendor and model - vendorid = db.Column( - db.Integer, - db.ForeignKey('vendors.vendorid'), - nullable=True - ) - modelnumberid = db.Column( - db.Integer, - db.ForeignKey('models.modelnumberid'), - nullable=True - ) - - # Equipment-specific fields - requiresmanualconfig = db.Column( - db.Boolean, - default=False, - comment='Multi-PC machine needs manual configuration' - ) - islocationonly = db.Column( - db.Boolean, - default=False, - comment='Virtual location marker (not actual equipment)' - ) - - # Maintenance tracking - lastmaintenancedate = db.Column(db.DateTime, nullable=True) - nextmaintenancedate = db.Column(db.DateTime, nullable=True) - maintenanceintervaldays = db.Column(db.Integer, nullable=True) - - # Controller info (for CNC machines) - controllervendorid = db.Column( - db.Integer, - db.ForeignKey('vendors.vendorid'), - nullable=True, - comment='Controller vendor (e.g., FANUC)' - ) - controllermodelid = db.Column( - db.Integer, - db.ForeignKey('models.modelnumberid'), - nullable=True, - comment='Controller model (e.g., 31B)' - ) - - # Relationships - asset = db.relationship( - 'Asset', - backref=db.backref('equipment', uselist=False, lazy='joined') - ) - equipmenttype = db.relationship('EquipmentType', backref='equipment') - vendor = db.relationship('Vendor', foreign_keys=[vendorid], backref='equipment_items') - model = db.relationship('Model', foreign_keys=[modelnumberid], backref='equipment_items') - controllervendor = db.relationship('Vendor', foreign_keys=[controllervendorid], backref='equipment_controllers') - controllermodel = db.relationship('Model', foreign_keys=[controllermodelid], backref='equipment_controller_models') - - __table_args__ = ( - db.Index('idx_equipment_type', 'equipmenttypeid'), - db.Index('idx_equipment_vendor', 'vendorid'), - ) - - def __repr__(self): - return f"" - - def to_dict(self): - """Convert to dictionary with related names.""" - result = super().to_dict() - - # Add related object names - if self.equipmenttype: - result['equipmenttypename'] = self.equipmenttype.equipmenttype - if self.vendor: - result['vendorname'] = self.vendor.vendor - if self.model: - result['modelname'] = self.model.modelnumber - if self.model.imageurl: - result['imageurl'] = self.model.imageurl - - # Add controller info - if self.controllervendor: - result['controllervendorname'] = self.controllervendor.vendor - if self.controllermodel: - result['controllermodelname'] = self.controllermodel.modelnumber - - return result +"""Equipment plugin models.""" + +from shopdb.api import db, BaseModel + + +class EquipmentType(BaseModel): + """ + Equipment type classification. + + Examples: CNC, CMM, Lathe, Grinder, EDM, Part Marker, etc. + """ + __tablename__ = 'equipmenttypes' + + equipmenttypeid = db.Column(db.Integer, primary_key=True) + equipmenttype = db.Column(db.String(100), unique=True, nullable=False) + description = db.Column(db.Text) + icon = db.Column(db.String(50), comment='Icon name for UI') + + def __repr__(self): + return f"" + + +class Equipment(BaseModel): + """ + Equipment-specific extension data. + + Links to core Asset table via assetid. + Stores equipment-specific fields like type, model, vendor, etc. + """ + __tablename__ = 'equipment' + + equipmentid = db.Column(db.Integer, primary_key=True) + + # Link to core asset + assetid = db.Column( + db.Integer, + db.ForeignKey('assets.assetid', ondelete='CASCADE'), + unique=True, + nullable=False, + index=True + ) + + # Equipment classification + equipmenttypeid = db.Column( + db.Integer, + db.ForeignKey('equipmenttypes.equipmenttypeid'), + nullable=True + ) + + # Vendor and model + vendorid = db.Column( + db.Integer, + db.ForeignKey('vendors.vendorid'), + nullable=True + ) + modelnumberid = db.Column( + db.Integer, + db.ForeignKey('models.modelnumberid'), + nullable=True + ) + + # Equipment-specific fields + requiresmanualconfig = db.Column( + db.Boolean, + default=False, + comment='Multi-PC machine needs manual configuration' + ) + islocationonly = db.Column( + db.Boolean, + default=False, + comment='Virtual location marker (not actual equipment)' + ) + + # Maintenance tracking + lastmaintenancedate = db.Column(db.DateTime, nullable=True) + nextmaintenancedate = db.Column(db.DateTime, nullable=True) + maintenanceintervaldays = db.Column(db.Integer, nullable=True) + + # Controller info (for CNC machines) + controllervendorid = db.Column( + db.Integer, + db.ForeignKey('vendors.vendorid'), + nullable=True, + comment='Controller vendor (e.g., FANUC)' + ) + controllermodelid = db.Column( + db.Integer, + db.ForeignKey('models.modelnumberid'), + nullable=True, + comment='Controller model (e.g., 31B)' + ) + + # Relationships + asset = db.relationship( + 'Asset', + backref=db.backref('equipment', uselist=False, lazy='joined') + ) + equipmenttype = db.relationship('EquipmentType', backref='equipment') + vendor = db.relationship('Vendor', foreign_keys=[vendorid], backref='equipment_items') + model = db.relationship('Model', foreign_keys=[modelnumberid], backref='equipment_items') + controllervendor = db.relationship('Vendor', foreign_keys=[controllervendorid], backref='equipment_controllers') + controllermodel = db.relationship('Model', foreign_keys=[controllermodelid], backref='equipment_controller_models') + + __table_args__ = ( + db.Index('idx_equipment_type', 'equipmenttypeid'), + db.Index('idx_equipment_vendor', 'vendorid'), + ) + + def __repr__(self): + return f"" + + def to_dict(self): + """Convert to dictionary with related names.""" + result = super().to_dict() + + # Add related object names + if self.equipmenttype: + result['equipmenttypename'] = self.equipmenttype.equipmenttype + if self.vendor: + result['vendorname'] = self.vendor.vendor + if self.model: + result['modelname'] = self.model.modelnumber + if self.model.imageurl: + result['imageurl'] = self.model.imageurl + + # Add controller info + if self.controllervendor: + result['controllervendorname'] = self.controllervendor.vendor + if self.controllermodel: + result['controllermodelname'] = self.controllermodel.modelnumber + + return result diff --git a/plugins/equipment/plugin.py b/plugins/equipment/plugin.py index bbe5b06..dc576a8 100644 --- a/plugins/equipment/plugin.py +++ b/plugins/equipment/plugin.py @@ -1,220 +1,219 @@ -"""Equipment 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 Equipment, EquipmentType -from .api import equipment_bp - -logger = logging.getLogger(__name__) - - -class EquipmentPlugin(BasePlugin): - """ - Equipment plugin - manages manufacturing equipment assets. - - Equipment includes CNCs, CMMs, lathes, grinders, EDMs, part markers, etc. - Uses the new Asset architecture with Equipment 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', 'equipment'), - version=self._manifest.get('version', '1.0.0'), - description=self._manifest.get( - 'description', - 'Equipment management for manufacturing assets' - ), - 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/equipment'), - ) - - def get_blueprint(self) -> Optional[Blueprint]: - """Return Flask Blueprint with API routes.""" - return equipment_bp - - def get_models(self) -> List[Type]: - """Return list of SQLAlchemy model classes.""" - return [Equipment, EquipmentType] - - def init_app(self, app: Flask, db_instance) -> None: - """Initialize plugin with Flask app.""" - logger.info(f"Equipment plugin initialized (v{self.meta.version})") - - def on_install(self, app: Flask) -> None: - """Called when plugin is installed.""" - with app.app_context(): - self._ensure_asset_type() - self._ensure_asset_statuses() - self._ensure_equipment_types() - logger.info("Equipment plugin installed") - - def _ensure_asset_type(self) -> None: - """Ensure equipment asset type exists.""" - existing = AssetType.query.filter_by(assettype='equipment').first() - if not existing: - at = AssetType( - assettype='equipment', - pluginname='equipment', - tablename='equipment', - description='Manufacturing equipment (CNCs, CMMs, lathes, etc.)', - icon='cog' - ) - db.session.add(at) - logger.debug("Created asset type: equipment") - db.session.commit() - - def _ensure_asset_statuses(self) -> None: - """Ensure standard asset statuses exist.""" - statuses = [ - ('In Use', 'Asset is currently in use', '#28a745'), - ('Spare', 'Spare/backup asset', '#17a2b8'), - ('Retired', 'Asset has been retired', '#6c757d'), - ('Maintenance', 'Asset is under maintenance', '#ffc107'), - ('Decommissioned', 'Asset has been decommissioned', '#dc3545'), - ] - - for name, description, color in statuses: - existing = AssetStatus.query.filter_by(status=name).first() - if not existing: - s = AssetStatus( - status=name, - description=description, - color=color - ) - db.session.add(s) - logger.debug(f"Created asset status: {name}") - - db.session.commit() - - def _ensure_equipment_types(self) -> None: - """Ensure basic equipment types exist.""" - equipment_types = [ - ('CNC', 'Computer Numerical Control machine', 'cnc'), - ('CMM', 'Coordinate Measuring Machine', 'cmm'), - ('Lathe', 'Lathe machine', 'lathe'), - ('Grinder', 'Grinding machine', 'grinder'), - ('EDM', 'Electrical Discharge Machine', 'edm'), - ('Part Marker', 'Part marking/engraving equipment', 'marker'), - ('Mill', 'Milling machine', 'mill'), - ('Press', 'Press machine', 'press'), - ('Robot', 'Industrial robot', 'robot'), - ('Other', 'Other equipment type', 'cog'), - ] - - for name, description, icon in equipment_types: - existing = EquipmentType.query.filter_by(equipmenttype=name).first() - if not existing: - et = EquipmentType( - equipmenttype=name, - description=description, - icon=icon - ) - db.session.add(et) - logger.debug(f"Created equipment type: {name}") - - db.session.commit() - - def on_uninstall(self, app: Flask) -> None: - """Called when plugin is uninstalled.""" - logger.info("Equipment plugin uninstalled") - - def get_cli_commands(self) -> List: - """Return CLI commands for this plugin.""" - - @click.group('equipment') - def equipmentcli(): - """Equipment plugin commands.""" - pass - - @equipmentcli.command('list-types') - def list_types(): - """List all equipment types.""" - from flask import current_app - - with current_app.app_context(): - types = EquipmentType.query.filter_by(isactive=True).all() - if not types: - click.echo('No equipment types found.') - return - - click.echo('Equipment Types:') - for t in types: - click.echo(f" [{t.equipmenttypeid}] {t.equipmenttype}") - - @equipmentcli.command('stats') - def stats(): - """Show equipment statistics.""" - from flask import current_app - from shopdb.core.models import Asset - - with current_app.app_context(): - total = db.session.query(Equipment).join(Asset).filter( - Asset.isactive == True - ).count() - - click.echo(f"Total active equipment: {total}") - - # By type - by_type = db.session.query( - EquipmentType.equipmenttype, - db.func.count(Equipment.equipmentid) - ).join(Equipment, Equipment.equipmenttypeid == EquipmentType.equipmenttypeid - ).join(Asset, Asset.assetid == Equipment.assetid - ).filter(Asset.isactive == True - ).group_by(EquipmentType.equipmenttype - ).all() - - if by_type: - click.echo("\nBy Type:") - for t, c in by_type: - click.echo(f" {t}: {c}") - - return [equipmentcli] - - def get_dashboard_widgets(self) -> List[Dict]: - """Return dashboard widget definitions.""" - return [ - { - 'name': 'Equipment Status', - 'component': 'EquipmentStatusWidget', - 'endpoint': '/api/equipment/dashboard/summary', - 'size': 'medium', - 'position': 5, - }, - ] - - def get_navigation_items(self) -> List[Dict]: - """Return navigation menu items.""" - return [ - { - 'name': 'Equipment', - 'icon': 'cog', - 'route': '/machines', - 'position': 10, - }, - ] +"""Equipment 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 Equipment, EquipmentType +from .api import equipment_bp + +logger = logging.getLogger(__name__) + + +class EquipmentPlugin(BasePlugin): + """ + Equipment plugin - manages manufacturing equipment assets. + + Equipment includes CNCs, CMMs, lathes, grinders, EDMs, part markers, etc. + Uses the new Asset architecture with Equipment 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', 'equipment'), + version=self._manifest.get('version', '1.0.0'), + description=self._manifest.get( + 'description', + 'Equipment management for manufacturing assets' + ), + 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/equipment'), + ) + + def get_blueprint(self) -> Optional[Blueprint]: + """Return Flask Blueprint with API routes.""" + return equipment_bp + + def get_models(self) -> List[Type]: + """Return list of SQLAlchemy model classes.""" + return [Equipment, EquipmentType] + + def init_app(self, app: Flask, db_instance) -> None: + """Initialize plugin with Flask app.""" + logger.info(f"Equipment plugin initialized (v{self.meta.version})") + + def on_install(self, app: Flask) -> None: + """Called when plugin is installed.""" + with app.app_context(): + self._ensure_asset_type() + self._ensure_asset_statuses() + self._ensure_equipment_types() + logger.info("Equipment plugin installed") + + def _ensure_asset_type(self) -> None: + """Ensure equipment asset type exists.""" + existing = AssetType.query.filter_by(assettype='equipment').first() + if not existing: + at = AssetType( + assettype='equipment', + pluginname='equipment', + tablename='equipment', + description='Manufacturing equipment (CNCs, CMMs, lathes, etc.)', + icon='cog' + ) + db.session.add(at) + logger.debug("Created asset type: equipment") + db.session.commit() + + def _ensure_asset_statuses(self) -> None: + """Ensure standard asset statuses exist.""" + statuses = [ + ('In Use', 'Asset is currently in use', '#28a745'), + ('Spare', 'Spare/backup asset', '#17a2b8'), + ('Retired', 'Asset has been retired', '#6c757d'), + ('Maintenance', 'Asset is under maintenance', '#ffc107'), + ('Decommissioned', 'Asset has been decommissioned', '#dc3545'), + ] + + for name, description, color in statuses: + existing = AssetStatus.query.filter_by(status=name).first() + if not existing: + s = AssetStatus( + status=name, + description=description, + color=color + ) + db.session.add(s) + logger.debug(f"Created asset status: {name}") + + db.session.commit() + + def _ensure_equipment_types(self) -> None: + """Ensure basic equipment types exist.""" + equipment_types = [ + ('CNC', 'Computer Numerical Control machine', 'cnc'), + ('CMM', 'Coordinate Measuring Machine', 'cmm'), + ('Lathe', 'Lathe machine', 'lathe'), + ('Grinder', 'Grinding machine', 'grinder'), + ('EDM', 'Electrical Discharge Machine', 'edm'), + ('Part Marker', 'Part marking/engraving equipment', 'marker'), + ('Mill', 'Milling machine', 'mill'), + ('Press', 'Press machine', 'press'), + ('Robot', 'Industrial robot', 'robot'), + ('Other', 'Other equipment type', 'cog'), + ] + + for name, description, icon in equipment_types: + existing = EquipmentType.query.filter_by(equipmenttype=name).first() + if not existing: + et = EquipmentType( + equipmenttype=name, + description=description, + icon=icon + ) + db.session.add(et) + logger.debug(f"Created equipment type: {name}") + + db.session.commit() + + def on_uninstall(self, app: Flask) -> None: + """Called when plugin is uninstalled.""" + logger.info("Equipment plugin uninstalled") + + def get_cli_commands(self) -> List: + """Return CLI commands for this plugin.""" + + @click.group('equipment') + def equipmentcli(): + """Equipment plugin commands.""" + pass + + @equipmentcli.command('list-types') + def list_types(): + """List all equipment types.""" + from flask import current_app + + with current_app.app_context(): + types = EquipmentType.query.filter_by(isactive=True).all() + if not types: + click.echo('No equipment types found.') + return + + click.echo('Equipment Types:') + for t in types: + click.echo(f" [{t.equipmenttypeid}] {t.equipmenttype}") + + @equipmentcli.command('stats') + def stats(): + """Show equipment statistics.""" + from flask import current_app + from shopdb.api import Asset + + with current_app.app_context(): + total = db.session.query(Equipment).join(Asset).filter( + Asset.isactive == True + ).count() + + click.echo(f"Total active equipment: {total}") + + # By type + by_type = db.session.query( + EquipmentType.equipmenttype, + db.func.count(Equipment.equipmentid) + ).join(Equipment, Equipment.equipmenttypeid == EquipmentType.equipmenttypeid + ).join(Asset, Asset.assetid == Equipment.assetid + ).filter(Asset.isactive == True + ).group_by(EquipmentType.equipmenttype + ).all() + + if by_type: + click.echo("\nBy Type:") + for t, c in by_type: + click.echo(f" {t}: {c}") + + return [equipmentcli] + + def get_dashboard_widgets(self) -> List[Dict]: + """Return dashboard widget definitions.""" + return [ + { + 'name': 'Equipment Status', + 'component': 'EquipmentStatusWidget', + 'endpoint': '/api/equipment/dashboard/summary', + 'size': 'medium', + 'position': 5, + }, + ] + + def get_navigation_items(self) -> List[Dict]: + """Return navigation menu items.""" + return [ + { + 'name': 'Equipment', + 'icon': 'cog', + 'route': '/machines', + 'position': 10, + }, + ] diff --git a/plugins/network/api/routes.py b/plugins/network/api/routes.py index 40e1113..649d0ee 100644 --- a/plugins/network/api/routes.py +++ b/plugins/network/api/routes.py @@ -3,15 +3,7 @@ from flask import Blueprint, request from flask_jwt_extended import jwt_required -from shopdb.extensions import db -from shopdb.core.models import Asset, AssetType, Vendor, AuditLog -from shopdb.utils.responses import ( - success_response, - error_response, - paginated_response, - ErrorCodes -) -from shopdb.utils.pagination import get_pagination_params, paginate_query +from shopdb.api import db, Asset, AssetType, Vendor, AuditLog, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query from ..models import NetworkDevice, NetworkDeviceType, Subnet, VLAN diff --git a/plugins/network/models/network_device.py b/plugins/network/models/network_device.py index da8ac04..c4a8490 100644 --- a/plugins/network/models/network_device.py +++ b/plugins/network/models/network_device.py @@ -1,121 +1,120 @@ -"""Network device plugin models.""" - -from shopdb.extensions import db -from shopdb.core.models.base import BaseModel - - -class NetworkDeviceType(BaseModel): - """ - Network device type classification. - - Examples: Switch, Router, Access Point, Camera, IDF, Firewall, etc. - """ - __tablename__ = 'networkdevicetypes' - - networkdevicetypeid = db.Column(db.Integer, primary_key=True) - networkdevicetype = db.Column(db.String(100), unique=True, nullable=False) - description = db.Column(db.Text) - icon = db.Column(db.String(50), comment='Icon name for UI') - - def __repr__(self): - return f"" - - -class NetworkDevice(BaseModel): - """ - Network device-specific extension data. - - Links to core Asset table via assetid. - Stores network device-specific fields like hostname, firmware, ports, etc. - """ - __tablename__ = 'networkdevices' - - networkdeviceid = db.Column(db.Integer, primary_key=True) - - # Link to core asset - assetid = db.Column( - db.Integer, - db.ForeignKey('assets.assetid', ondelete='CASCADE'), - unique=True, - nullable=False, - index=True - ) - - # Network device classification - networkdevicetypeid = db.Column( - db.Integer, - db.ForeignKey('networkdevicetypes.networkdevicetypeid'), - nullable=True - ) - - # Vendor - vendorid = db.Column( - db.Integer, - db.ForeignKey('vendors.vendorid'), - nullable=True - ) - - # Network identity - hostname = db.Column( - db.String(100), - index=True, - comment='Network hostname' - ) - - # Firmware/software version - firmwareversion = db.Column(db.String(100), nullable=True) - - # Physical characteristics - portcount = db.Column( - db.Integer, - nullable=True, - comment='Number of ports (for switches)' - ) - - # Features - ispoe = db.Column( - db.Boolean, - default=False, - comment='Power over Ethernet capable' - ) - ismanaged = db.Column( - db.Boolean, - default=False, - comment='Managed device (SNMP, web interface, etc.)' - ) - - # For IDF/closet locations - rackunit = db.Column( - db.String(20), - nullable=True, - comment='Rack unit position (e.g., U1, U5)' - ) - - # Relationships - asset = db.relationship( - 'Asset', - backref=db.backref('network_device', uselist=False, lazy='joined') - ) - networkdevicetype = db.relationship('NetworkDeviceType', backref='networkdevices') - vendor = db.relationship('Vendor', backref='network_devices') - - __table_args__ = ( - db.Index('idx_netdev_type', 'networkdevicetypeid'), - db.Index('idx_netdev_hostname', 'hostname'), - db.Index('idx_netdev_vendor', 'vendorid'), - ) - - def __repr__(self): - return f"" - - def to_dict(self): - """Convert to dictionary with related names.""" - result = super().to_dict() - - # Add related object names - if self.networkdevicetype: - result['networkdevicetypename'] = self.networkdevicetype.networkdevicetype - if self.vendor: - result['vendorname'] = self.vendor.vendor - - return result +"""Network device plugin models.""" + +from shopdb.api import db, BaseModel + + +class NetworkDeviceType(BaseModel): + """ + Network device type classification. + + Examples: Switch, Router, Access Point, Camera, IDF, Firewall, etc. + """ + __tablename__ = 'networkdevicetypes' + + networkdevicetypeid = db.Column(db.Integer, primary_key=True) + networkdevicetype = db.Column(db.String(100), unique=True, nullable=False) + description = db.Column(db.Text) + icon = db.Column(db.String(50), comment='Icon name for UI') + + def __repr__(self): + return f"" + + +class NetworkDevice(BaseModel): + """ + Network device-specific extension data. + + Links to core Asset table via assetid. + Stores network device-specific fields like hostname, firmware, ports, etc. + """ + __tablename__ = 'networkdevices' + + networkdeviceid = db.Column(db.Integer, primary_key=True) + + # Link to core asset + assetid = db.Column( + db.Integer, + db.ForeignKey('assets.assetid', ondelete='CASCADE'), + unique=True, + nullable=False, + index=True + ) + + # Network device classification + networkdevicetypeid = db.Column( + db.Integer, + db.ForeignKey('networkdevicetypes.networkdevicetypeid'), + nullable=True + ) + + # Vendor + vendorid = db.Column( + db.Integer, + db.ForeignKey('vendors.vendorid'), + nullable=True + ) + + # Network identity + hostname = db.Column( + db.String(100), + index=True, + comment='Network hostname' + ) + + # Firmware/software version + firmwareversion = db.Column(db.String(100), nullable=True) + + # Physical characteristics + portcount = db.Column( + db.Integer, + nullable=True, + comment='Number of ports (for switches)' + ) + + # Features + ispoe = db.Column( + db.Boolean, + default=False, + comment='Power over Ethernet capable' + ) + ismanaged = db.Column( + db.Boolean, + default=False, + comment='Managed device (SNMP, web interface, etc.)' + ) + + # For IDF/closet locations + rackunit = db.Column( + db.String(20), + nullable=True, + comment='Rack unit position (e.g., U1, U5)' + ) + + # Relationships + asset = db.relationship( + 'Asset', + backref=db.backref('network_device', uselist=False, lazy='joined') + ) + networkdevicetype = db.relationship('NetworkDeviceType', backref='networkdevices') + vendor = db.relationship('Vendor', backref='network_devices') + + __table_args__ = ( + db.Index('idx_netdev_type', 'networkdevicetypeid'), + db.Index('idx_netdev_hostname', 'hostname'), + db.Index('idx_netdev_vendor', 'vendorid'), + ) + + def __repr__(self): + return f"" + + def to_dict(self): + """Convert to dictionary with related names.""" + result = super().to_dict() + + # Add related object names + if self.networkdevicetype: + result['networkdevicetypename'] = self.networkdevicetype.networkdevicetype + if self.vendor: + result['vendorname'] = self.vendor.vendor + + return result diff --git a/plugins/network/models/subnet.py b/plugins/network/models/subnet.py index e2332f2..6a52b9f 100644 --- a/plugins/network/models/subnet.py +++ b/plugins/network/models/subnet.py @@ -1,146 +1,145 @@ -"""Subnet and VLAN models for network plugin.""" - -from shopdb.extensions import db -from shopdb.core.models.base import BaseModel - - -class VLAN(BaseModel): - """ - VLAN definition. - - Represents a virtual LAN for network segmentation. - """ - __tablename__ = 'vlans' - - vlanid = db.Column(db.Integer, primary_key=True) - vlannumber = db.Column(db.Integer, unique=True, nullable=False, comment='VLAN ID number') - name = db.Column(db.String(100), nullable=False, comment='VLAN name') - description = db.Column(db.Text, nullable=True) - - # Optional classification - vlantype = db.Column( - db.String(50), - nullable=True, - comment='Type: data, voice, management, guest, etc.' - ) - - # Relationships - subnets = db.relationship('Subnet', backref='vlan', lazy='dynamic') - - __table_args__ = ( - db.Index('idx_vlan_number', 'vlannumber'), - ) - - def __repr__(self): - return f"" - - def to_dict(self): - """Convert to dictionary.""" - result = super().to_dict() - result['subnetcount'] = self.subnets.count() if self.subnets else 0 - return result - - -class Subnet(BaseModel): - """ - Subnet/IP network definition. - - Represents an IP subnet with optional VLAN association. - """ - __tablename__ = 'subnets' - - subnetid = db.Column(db.Integer, primary_key=True) - - # Network definition - cidr = db.Column( - db.String(18), - unique=True, - nullable=False, - comment='CIDR notation (e.g., 10.1.1.0/24)' - ) - name = db.Column(db.String(100), nullable=False, comment='Subnet name') - description = db.Column(db.Text, nullable=True) - - # Network details - gatewayip = db.Column( - db.String(15), - nullable=True, - comment='Default gateway IP address' - ) - subnetmask = db.Column( - db.String(15), - nullable=True, - comment='Subnet mask (e.g., 255.255.255.0)' - ) - networkaddress = db.Column( - db.String(15), - nullable=True, - comment='Network address (e.g., 10.1.1.0)' - ) - broadcastaddress = db.Column( - db.String(15), - nullable=True, - comment='Broadcast address (e.g., 10.1.1.255)' - ) - - # VLAN association - vlanid = db.Column( - db.Integer, - db.ForeignKey('vlans.vlanid'), - nullable=True - ) - - # Classification - subnettype = db.Column( - db.String(50), - nullable=True, - comment='Type: production, development, management, dmz, etc.' - ) - - # Location association - locationid = db.Column( - db.Integer, - db.ForeignKey('locations.locationid'), - nullable=True - ) - - # DHCP settings - dhcpenabled = db.Column(db.Boolean, default=True, comment='DHCP enabled for this subnet') - dhcprangestart = db.Column(db.String(15), nullable=True, comment='DHCP range start IP') - dhcprangeend = db.Column(db.String(15), nullable=True, comment='DHCP range end IP') - - # DNS settings - dns1 = db.Column(db.String(15), nullable=True, comment='Primary DNS server') - dns2 = db.Column(db.String(15), nullable=True, comment='Secondary DNS server') - - # Relationships - location = db.relationship('Location', backref='subnets') - - __table_args__ = ( - db.Index('idx_subnet_cidr', 'cidr'), - db.Index('idx_subnet_vlan', 'vlanid'), - db.Index('idx_subnet_location', 'locationid'), - ) - - def __repr__(self): - return f"" - - @property - def vlan_number(self): - """Get the VLAN number.""" - return self.vlan.vlannumber if self.vlan else None - - def to_dict(self): - """Convert to dictionary with related data.""" - result = super().to_dict() - - # Add VLAN info - if self.vlan: - result['vlannumber'] = self.vlan.vlannumber - result['vlanname'] = self.vlan.name - - # Add location info - if self.location: - result['locationname'] = self.location.locationname - - return result +"""Subnet and VLAN models for network plugin.""" + +from shopdb.api import db, BaseModel + + +class VLAN(BaseModel): + """ + VLAN definition. + + Represents a virtual LAN for network segmentation. + """ + __tablename__ = 'vlans' + + vlanid = db.Column(db.Integer, primary_key=True) + vlannumber = db.Column(db.Integer, unique=True, nullable=False, comment='VLAN ID number') + name = db.Column(db.String(100), nullable=False, comment='VLAN name') + description = db.Column(db.Text, nullable=True) + + # Optional classification + vlantype = db.Column( + db.String(50), + nullable=True, + comment='Type: data, voice, management, guest, etc.' + ) + + # Relationships + subnets = db.relationship('Subnet', backref='vlan', lazy='dynamic') + + __table_args__ = ( + db.Index('idx_vlan_number', 'vlannumber'), + ) + + def __repr__(self): + return f"" + + def to_dict(self): + """Convert to dictionary.""" + result = super().to_dict() + result['subnetcount'] = self.subnets.count() if self.subnets else 0 + return result + + +class Subnet(BaseModel): + """ + Subnet/IP network definition. + + Represents an IP subnet with optional VLAN association. + """ + __tablename__ = 'subnets' + + subnetid = db.Column(db.Integer, primary_key=True) + + # Network definition + cidr = db.Column( + db.String(18), + unique=True, + nullable=False, + comment='CIDR notation (e.g., 10.1.1.0/24)' + ) + name = db.Column(db.String(100), nullable=False, comment='Subnet name') + description = db.Column(db.Text, nullable=True) + + # Network details + gatewayip = db.Column( + db.String(15), + nullable=True, + comment='Default gateway IP address' + ) + subnetmask = db.Column( + db.String(15), + nullable=True, + comment='Subnet mask (e.g., 255.255.255.0)' + ) + networkaddress = db.Column( + db.String(15), + nullable=True, + comment='Network address (e.g., 10.1.1.0)' + ) + broadcastaddress = db.Column( + db.String(15), + nullable=True, + comment='Broadcast address (e.g., 10.1.1.255)' + ) + + # VLAN association + vlanid = db.Column( + db.Integer, + db.ForeignKey('vlans.vlanid'), + nullable=True + ) + + # Classification + subnettype = db.Column( + db.String(50), + nullable=True, + comment='Type: production, development, management, dmz, etc.' + ) + + # Location association + locationid = db.Column( + db.Integer, + db.ForeignKey('locations.locationid'), + nullable=True + ) + + # DHCP settings + dhcpenabled = db.Column(db.Boolean, default=True, comment='DHCP enabled for this subnet') + dhcprangestart = db.Column(db.String(15), nullable=True, comment='DHCP range start IP') + dhcprangeend = db.Column(db.String(15), nullable=True, comment='DHCP range end IP') + + # DNS settings + dns1 = db.Column(db.String(15), nullable=True, comment='Primary DNS server') + dns2 = db.Column(db.String(15), nullable=True, comment='Secondary DNS server') + + # Relationships + location = db.relationship('Location', backref='subnets') + + __table_args__ = ( + db.Index('idx_subnet_cidr', 'cidr'), + db.Index('idx_subnet_vlan', 'vlanid'), + db.Index('idx_subnet_location', 'locationid'), + ) + + def __repr__(self): + return f"" + + @property + def vlan_number(self): + """Get the VLAN number.""" + return self.vlan.vlannumber if self.vlan else None + + def to_dict(self): + """Convert to dictionary with related data.""" + result = super().to_dict() + + # Add VLAN info + if self.vlan: + result['vlannumber'] = self.vlan.vlannumber + result['vlanname'] = self.vlan.name + + # Add location info + if self.location: + result['locationname'] = self.location.locationname + + return result diff --git a/plugins/network/plugin.py b/plugins/network/plugin.py index 1ca063d..4694ecd 100644 --- a/plugins/network/plugin.py +++ b/plugins/network/plugin.py @@ -1,217 +1,216 @@ -"""Network 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 - -from .models import NetworkDevice, NetworkDeviceType, Subnet, VLAN -from .api import network_bp - -logger = logging.getLogger(__name__) - - -class NetworkPlugin(BasePlugin): - """ - Network plugin - manages network device assets. - - Network devices include switches, routers, access points, cameras, IDFs, etc. - Uses the new Asset architecture with NetworkDevice 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', 'network'), - version=self._manifest.get('version', '1.0.0'), - description=self._manifest.get( - 'description', - 'Network device management for switches, APs, and cameras' - ), - 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/network'), - ) - - def get_blueprint(self) -> Optional[Blueprint]: - """Return Flask Blueprint with API routes.""" - return network_bp - - def get_models(self) -> List[Type]: - """Return list of SQLAlchemy model classes.""" - return [NetworkDevice, NetworkDeviceType, Subnet, VLAN] - - def init_app(self, app: Flask, db_instance) -> None: - """Initialize plugin with Flask app.""" - logger.info(f"Network plugin initialized (v{self.meta.version})") - - def on_install(self, app: Flask) -> None: - """Called when plugin is installed.""" - with app.app_context(): - self._ensure_asset_type() - self._ensure_network_device_types() - logger.info("Network plugin installed") - - def _ensure_asset_type(self) -> None: - """Ensure network_device asset type exists.""" - existing = AssetType.query.filter_by(assettype='network_device').first() - if not existing: - at = AssetType( - assettype='network_device', - pluginname='network', - tablename='networkdevices', - description='Network infrastructure devices (switches, APs, cameras, etc.)', - icon='network-wired' - ) - db.session.add(at) - logger.debug("Created asset type: network_device") - db.session.commit() - - def _ensure_network_device_types(self) -> None: - """Ensure basic network device types exist.""" - device_types = [ - ('Switch', 'Network switch', 'network-wired'), - ('Router', 'Network router', 'router'), - ('Access Point', 'Wireless access point', 'wifi'), - ('Firewall', 'Network firewall', 'shield'), - ('Camera', 'IP camera', 'video'), - ('IDF', 'Intermediate Distribution Frame/closet', 'box'), - ('MDF', 'Main Distribution Frame', 'building'), - ('Patch Panel', 'Patch panel', 'th'), - ('UPS', 'Uninterruptible power supply', 'battery'), - ('Other', 'Other network device', 'network-wired'), - ] - - for name, description, icon in device_types: - existing = NetworkDeviceType.query.filter_by(networkdevicetype=name).first() - if not existing: - ndt = NetworkDeviceType( - networkdevicetype=name, - description=description, - icon=icon - ) - db.session.add(ndt) - logger.debug(f"Created network device type: {name}") - - db.session.commit() - - def on_uninstall(self, app: Flask) -> None: - """Called when plugin is uninstalled.""" - logger.info("Network plugin uninstalled") - - def get_cli_commands(self) -> List: - """Return CLI commands for this plugin.""" - - @click.group('network') - def networkcli(): - """Network plugin commands.""" - pass - - @networkcli.command('list-types') - def list_types(): - """List all network device types.""" - from flask import current_app - - with current_app.app_context(): - types = NetworkDeviceType.query.filter_by(isactive=True).all() - if not types: - click.echo('No network device types found.') - return - - click.echo('Network Device Types:') - for t in types: - click.echo(f" [{t.networkdevicetypeid}] {t.networkdevicetype}") - - @networkcli.command('stats') - def stats(): - """Show network device statistics.""" - from flask import current_app - from shopdb.core.models import Asset - - with current_app.app_context(): - total = db.session.query(NetworkDevice).join(Asset).filter( - Asset.isactive == True - ).count() - - click.echo(f"Total active network devices: {total}") - - # By type - by_type = db.session.query( - NetworkDeviceType.networkdevicetype, - db.func.count(NetworkDevice.networkdeviceid) - ).join(NetworkDevice, NetworkDevice.networkdevicetypeid == NetworkDeviceType.networkdevicetypeid - ).join(Asset, Asset.assetid == NetworkDevice.assetid - ).filter(Asset.isactive == True - ).group_by(NetworkDeviceType.networkdevicetype - ).all() - - if by_type: - click.echo("\nBy Type:") - for t, c in by_type: - click.echo(f" {t}: {c}") - - @networkcli.command('find') - @click.argument('hostname') - def find_by_hostname(hostname): - """Find a network device by hostname.""" - from flask import current_app - - with current_app.app_context(): - netdev = NetworkDevice.query.filter( - NetworkDevice.hostname.ilike(f'%{hostname}%') - ).first() - - if not netdev: - click.echo(f'No network device found matching hostname: {hostname}') - return - - click.echo(f'Found: {netdev.hostname}') - click.echo(f' Asset: {netdev.asset.assetnumber}') - click.echo(f' Type: {netdev.networkdevicetype.networkdevicetype if netdev.networkdevicetype else "N/A"}') - click.echo(f' Firmware: {netdev.firmwareversion or "N/A"}') - click.echo(f' PoE: {"Yes" if netdev.ispoe else "No"}') - - return [networkcli] - - def get_dashboard_widgets(self) -> List[Dict]: - """Return dashboard widget definitions.""" - return [ - { - 'name': 'Network Status', - 'component': 'NetworkStatusWidget', - 'endpoint': '/api/network/dashboard/summary', - 'size': 'medium', - 'position': 7, - }, - ] - - def get_navigation_items(self) -> List[Dict]: - """Return navigation menu items.""" - return [ - { - 'name': 'Network', - 'icon': 'network-wired', - 'route': '/network', - 'position': 18, - }, - ] +"""Network 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 + +from .models import NetworkDevice, NetworkDeviceType, Subnet, VLAN +from .api import network_bp + +logger = logging.getLogger(__name__) + + +class NetworkPlugin(BasePlugin): + """ + Network plugin - manages network device assets. + + Network devices include switches, routers, access points, cameras, IDFs, etc. + Uses the new Asset architecture with NetworkDevice 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', 'network'), + version=self._manifest.get('version', '1.0.0'), + description=self._manifest.get( + 'description', + 'Network device management for switches, APs, and cameras' + ), + 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/network'), + ) + + def get_blueprint(self) -> Optional[Blueprint]: + """Return Flask Blueprint with API routes.""" + return network_bp + + def get_models(self) -> List[Type]: + """Return list of SQLAlchemy model classes.""" + return [NetworkDevice, NetworkDeviceType, Subnet, VLAN] + + def init_app(self, app: Flask, db_instance) -> None: + """Initialize plugin with Flask app.""" + logger.info(f"Network plugin initialized (v{self.meta.version})") + + def on_install(self, app: Flask) -> None: + """Called when plugin is installed.""" + with app.app_context(): + self._ensure_asset_type() + self._ensure_network_device_types() + logger.info("Network plugin installed") + + def _ensure_asset_type(self) -> None: + """Ensure network_device asset type exists.""" + existing = AssetType.query.filter_by(assettype='network_device').first() + if not existing: + at = AssetType( + assettype='network_device', + pluginname='network', + tablename='networkdevices', + description='Network infrastructure devices (switches, APs, cameras, etc.)', + icon='network-wired' + ) + db.session.add(at) + logger.debug("Created asset type: network_device") + db.session.commit() + + def _ensure_network_device_types(self) -> None: + """Ensure basic network device types exist.""" + device_types = [ + ('Switch', 'Network switch', 'network-wired'), + ('Router', 'Network router', 'router'), + ('Access Point', 'Wireless access point', 'wifi'), + ('Firewall', 'Network firewall', 'shield'), + ('Camera', 'IP camera', 'video'), + ('IDF', 'Intermediate Distribution Frame/closet', 'box'), + ('MDF', 'Main Distribution Frame', 'building'), + ('Patch Panel', 'Patch panel', 'th'), + ('UPS', 'Uninterruptible power supply', 'battery'), + ('Other', 'Other network device', 'network-wired'), + ] + + for name, description, icon in device_types: + existing = NetworkDeviceType.query.filter_by(networkdevicetype=name).first() + if not existing: + ndt = NetworkDeviceType( + networkdevicetype=name, + description=description, + icon=icon + ) + db.session.add(ndt) + logger.debug(f"Created network device type: {name}") + + db.session.commit() + + def on_uninstall(self, app: Flask) -> None: + """Called when plugin is uninstalled.""" + logger.info("Network plugin uninstalled") + + def get_cli_commands(self) -> List: + """Return CLI commands for this plugin.""" + + @click.group('network') + def networkcli(): + """Network plugin commands.""" + pass + + @networkcli.command('list-types') + def list_types(): + """List all network device types.""" + from flask import current_app + + with current_app.app_context(): + types = NetworkDeviceType.query.filter_by(isactive=True).all() + if not types: + click.echo('No network device types found.') + return + + click.echo('Network Device Types:') + for t in types: + click.echo(f" [{t.networkdevicetypeid}] {t.networkdevicetype}") + + @networkcli.command('stats') + def stats(): + """Show network device statistics.""" + from flask import current_app + from shopdb.api import Asset + + with current_app.app_context(): + total = db.session.query(NetworkDevice).join(Asset).filter( + Asset.isactive == True + ).count() + + click.echo(f"Total active network devices: {total}") + + # By type + by_type = db.session.query( + NetworkDeviceType.networkdevicetype, + db.func.count(NetworkDevice.networkdeviceid) + ).join(NetworkDevice, NetworkDevice.networkdevicetypeid == NetworkDeviceType.networkdevicetypeid + ).join(Asset, Asset.assetid == NetworkDevice.assetid + ).filter(Asset.isactive == True + ).group_by(NetworkDeviceType.networkdevicetype + ).all() + + if by_type: + click.echo("\nBy Type:") + for t, c in by_type: + click.echo(f" {t}: {c}") + + @networkcli.command('find') + @click.argument('hostname') + def find_by_hostname(hostname): + """Find a network device by hostname.""" + from flask import current_app + + with current_app.app_context(): + netdev = NetworkDevice.query.filter( + NetworkDevice.hostname.ilike(f'%{hostname}%') + ).first() + + if not netdev: + click.echo(f'No network device found matching hostname: {hostname}') + return + + click.echo(f'Found: {netdev.hostname}') + click.echo(f' Asset: {netdev.asset.assetnumber}') + click.echo(f' Type: {netdev.networkdevicetype.networkdevicetype if netdev.networkdevicetype else "N/A"}') + click.echo(f' Firmware: {netdev.firmwareversion or "N/A"}') + click.echo(f' PoE: {"Yes" if netdev.ispoe else "No"}') + + return [networkcli] + + def get_dashboard_widgets(self) -> List[Dict]: + """Return dashboard widget definitions.""" + return [ + { + 'name': 'Network Status', + 'component': 'NetworkStatusWidget', + 'endpoint': '/api/network/dashboard/summary', + 'size': 'medium', + 'position': 7, + }, + ] + + def get_navigation_items(self) -> List[Dict]: + """Return navigation menu items.""" + return [ + { + 'name': 'Network', + 'icon': 'network-wired', + 'route': '/network', + 'position': 18, + }, + ] diff --git a/plugins/notifications/api/routes.py b/plugins/notifications/api/routes.py index 4b738b6..8d43013 100644 --- a/plugins/notifications/api/routes.py +++ b/plugins/notifications/api/routes.py @@ -4,15 +4,7 @@ from datetime import datetime from flask import Blueprint, request from flask_jwt_extended import jwt_required -from shopdb.extensions import db -from shopdb.utils.responses import ( - success_response, - error_response, - paginated_response, - ErrorCodes -) -from shopdb.utils.pagination import get_pagination_params, paginate_query -from shopdb.utils.employee_db import employee_connection +from shopdb.api import db, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query, employee_connection from ..models import Notification, NotificationType diff --git a/plugins/notifications/models/notification.py b/plugins/notifications/models/notification.py index bcccd38..214bdc5 100644 --- a/plugins/notifications/models/notification.py +++ b/plugins/notifications/models/notification.py @@ -1,157 +1,157 @@ -"""Notifications plugin models - adapted to existing database schema.""" - -from datetime import datetime -from shopdb.extensions import db - - -class NotificationType(db.Model): - """ - Notification type classification. - Matches existing notificationtypes table. - """ - __tablename__ = 'notificationtypes' - - notificationtypeid = db.Column(db.Integer, primary_key=True) - typename = db.Column(db.String(50), nullable=False) - typedescription = db.Column(db.Text) - typecolor = db.Column(db.String(20), default='#17a2b8') - isactive = db.Column(db.Boolean, default=True) - - def __repr__(self): - return f"" - - def to_dict(self): - return { - 'notificationtypeid': self.notificationtypeid, - 'typename': self.typename, - 'typedescription': self.typedescription, - 'typecolor': self.typecolor, - 'isactive': self.isactive - } - - -class Notification(db.Model): - """ - Notification/announcement model. - Matches existing notifications table schema. - """ - __tablename__ = 'notifications' - - notificationid = db.Column(db.Integer, primary_key=True) - notificationtypeid = db.Column( - db.Integer, - db.ForeignKey('notificationtypes.notificationtypeid'), - nullable=True - ) - businessunitid = db.Column(db.Integer, nullable=True) - appid = db.Column(db.Integer, nullable=True) - notification = db.Column(db.Text, nullable=False, comment='The message content') - starttime = db.Column(db.DateTime, nullable=True) - endtime = db.Column(db.DateTime, nullable=True) - ticketnumber = db.Column(db.String(50), nullable=True) - link = db.Column(db.String(500), nullable=True) - isactive = db.Column(db.Boolean, default=True) - isshopfloor = db.Column(db.Boolean, default=False) - employeesso = db.Column(db.String(100), nullable=True) - employeename = db.Column(db.String(100), nullable=True) - - # Relationships - notificationtype = db.relationship('NotificationType', backref='notifications') - - def __repr__(self): - return f"" - - @property - def is_current(self): - """Check if notification is currently active based on dates.""" - now = datetime.utcnow() - if not self.isactive: - return False - if self.starttime and now < self.starttime: - return False - if self.endtime and now > self.endtime: - return False - return True - - @property - def title(self): - """Get title - first line or first 100 chars of notification.""" - if not self.notification: - return '' - lines = self.notification.split('\n') - return lines[0][:100] if lines else self.notification[:100] - - def to_dict(self): - """Convert to dictionary with related data.""" - result = { - 'notificationid': self.notificationid, - 'notificationtypeid': self.notificationtypeid, - 'businessunitid': self.businessunitid, - 'appid': self.appid, - 'notification': self.notification, - 'title': self.title, - 'message': self.notification, - 'starttime': self.starttime.isoformat() if self.starttime else None, - 'endtime': self.endtime.isoformat() if self.endtime else None, - 'startdate': self.starttime.isoformat() if self.starttime else None, - 'enddate': self.endtime.isoformat() if self.endtime else None, - 'ticketnumber': self.ticketnumber, - 'link': self.link, - 'linkurl': self.link, - 'isactive': bool(self.isactive) if self.isactive is not None else True, - 'isshopfloor': bool(self.isshopfloor) if self.isshopfloor is not None else False, - 'employeesso': self.employeesso, - 'employeename': self.employeename, - 'iscurrent': self.is_current - } - - # Add type info - if self.notificationtype: - result['typename'] = self.notificationtype.typename - result['typecolor'] = self.notificationtype.typecolor - - return result - - def to_calendar_event(self): - """Convert to FullCalendar event format.""" - # Map Bootstrap color names to hex colors - color_map = { - 'success': '#04b962', - 'warning': '#ff8800', - 'danger': '#f5365c', - 'info': '#14abef', - 'primary': '#7934f3', - 'secondary': '#94614f', - 'recognition': '#14abef', # Blue for recognition - } - - raw_color = self.notificationtype.typecolor if self.notificationtype else 'info' - # Use mapped color if it's a Bootstrap name, otherwise use as-is (hex) - color = color_map.get(raw_color, raw_color if raw_color.startswith('#') else '#14abef') - - # For recognition notifications, include employee name (or SSO as fallback) in title - title = self.title - if raw_color == 'recognition': - employee_display = self.employeename or self.employeesso - if employee_display: - title = f"{employee_display}: {title}" - - return { - 'id': self.notificationid, - 'title': title, - 'start': self.starttime.isoformat() if self.starttime else None, - 'end': self.endtime.isoformat() if self.endtime else None, - 'allDay': True, - 'backgroundColor': color, - 'borderColor': color, - 'extendedProps': { - 'notificationid': self.notificationid, - 'message': self.notification, - 'typename': self.notificationtype.typename if self.notificationtype else None, - 'typecolor': raw_color, - 'linkurl': self.link, - 'ticketnumber': self.ticketnumber, - 'employeename': self.employeename, - 'employeesso': self.employeesso, - } - } +"""Notifications plugin models - adapted to existing database schema.""" + +from datetime import datetime +from shopdb.api import db + + +class NotificationType(db.Model): + """ + Notification type classification. + Matches existing notificationtypes table. + """ + __tablename__ = 'notificationtypes' + + notificationtypeid = db.Column(db.Integer, primary_key=True) + typename = db.Column(db.String(50), nullable=False) + typedescription = db.Column(db.Text) + typecolor = db.Column(db.String(20), default='#17a2b8') + isactive = db.Column(db.Boolean, default=True) + + def __repr__(self): + return f"" + + def to_dict(self): + return { + 'notificationtypeid': self.notificationtypeid, + 'typename': self.typename, + 'typedescription': self.typedescription, + 'typecolor': self.typecolor, + 'isactive': self.isactive + } + + +class Notification(db.Model): + """ + Notification/announcement model. + Matches existing notifications table schema. + """ + __tablename__ = 'notifications' + + notificationid = db.Column(db.Integer, primary_key=True) + notificationtypeid = db.Column( + db.Integer, + db.ForeignKey('notificationtypes.notificationtypeid'), + nullable=True + ) + businessunitid = db.Column(db.Integer, nullable=True) + appid = db.Column(db.Integer, nullable=True) + notification = db.Column(db.Text, nullable=False, comment='The message content') + starttime = db.Column(db.DateTime, nullable=True) + endtime = db.Column(db.DateTime, nullable=True) + ticketnumber = db.Column(db.String(50), nullable=True) + link = db.Column(db.String(500), nullable=True) + isactive = db.Column(db.Boolean, default=True) + isshopfloor = db.Column(db.Boolean, default=False) + employeesso = db.Column(db.String(100), nullable=True) + employeename = db.Column(db.String(100), nullable=True) + + # Relationships + notificationtype = db.relationship('NotificationType', backref='notifications') + + def __repr__(self): + return f"" + + @property + def is_current(self): + """Check if notification is currently active based on dates.""" + now = datetime.utcnow() + if not self.isactive: + return False + if self.starttime and now < self.starttime: + return False + if self.endtime and now > self.endtime: + return False + return True + + @property + def title(self): + """Get title - first line or first 100 chars of notification.""" + if not self.notification: + return '' + lines = self.notification.split('\n') + return lines[0][:100] if lines else self.notification[:100] + + def to_dict(self): + """Convert to dictionary with related data.""" + result = { + 'notificationid': self.notificationid, + 'notificationtypeid': self.notificationtypeid, + 'businessunitid': self.businessunitid, + 'appid': self.appid, + 'notification': self.notification, + 'title': self.title, + 'message': self.notification, + 'starttime': self.starttime.isoformat() if self.starttime else None, + 'endtime': self.endtime.isoformat() if self.endtime else None, + 'startdate': self.starttime.isoformat() if self.starttime else None, + 'enddate': self.endtime.isoformat() if self.endtime else None, + 'ticketnumber': self.ticketnumber, + 'link': self.link, + 'linkurl': self.link, + 'isactive': bool(self.isactive) if self.isactive is not None else True, + 'isshopfloor': bool(self.isshopfloor) if self.isshopfloor is not None else False, + 'employeesso': self.employeesso, + 'employeename': self.employeename, + 'iscurrent': self.is_current + } + + # Add type info + if self.notificationtype: + result['typename'] = self.notificationtype.typename + result['typecolor'] = self.notificationtype.typecolor + + return result + + def to_calendar_event(self): + """Convert to FullCalendar event format.""" + # Map Bootstrap color names to hex colors + color_map = { + 'success': '#04b962', + 'warning': '#ff8800', + 'danger': '#f5365c', + 'info': '#14abef', + 'primary': '#7934f3', + 'secondary': '#94614f', + 'recognition': '#14abef', # Blue for recognition + } + + raw_color = self.notificationtype.typecolor if self.notificationtype else 'info' + # Use mapped color if it's a Bootstrap name, otherwise use as-is (hex) + color = color_map.get(raw_color, raw_color if raw_color.startswith('#') else '#14abef') + + # For recognition notifications, include employee name (or SSO as fallback) in title + title = self.title + if raw_color == 'recognition': + employee_display = self.employeename or self.employeesso + if employee_display: + title = f"{employee_display}: {title}" + + return { + 'id': self.notificationid, + 'title': title, + 'start': self.starttime.isoformat() if self.starttime else None, + 'end': self.endtime.isoformat() if self.endtime else None, + 'allDay': True, + 'backgroundColor': color, + 'borderColor': color, + 'extendedProps': { + 'notificationid': self.notificationid, + 'message': self.notification, + 'typename': self.notificationtype.typename if self.notificationtype else None, + 'typecolor': raw_color, + 'linkurl': self.link, + 'ticketnumber': self.ticketnumber, + 'employeename': self.employeename, + 'employeesso': self.employeesso, + } + } diff --git a/plugins/notifications/plugin.py b/plugins/notifications/plugin.py index 495a91a..cd7ef0a 100644 --- a/plugins/notifications/plugin.py +++ b/plugins/notifications/plugin.py @@ -1,204 +1,204 @@ -"""Notifications 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 .models import Notification, NotificationType -from .api import notifications_bp - -logger = logging.getLogger(__name__) - - -class NotificationsPlugin(BasePlugin): - """ - Notifications plugin - manages announcements and notifications. - - Provides functionality for: - - Creating and managing notifications/announcements - - Displaying banner notifications - - Calendar view of notifications - """ - - def __init__(self): - self._manifest = self._load_manifest() - - def _load_manifest(self) -> Dict: - """Load plugin manifest from JSON file.""" - manifest_path = Path(__file__).parent / 'manifest.json' - if manifest_path.exists(): - with open(manifest_path, 'r') as f: - return json.load(f) - return {} - - @property - def meta(self) -> PluginMeta: - """Return plugin metadata.""" - return PluginMeta( - name=self._manifest.get('name', 'notifications'), - version=self._manifest.get('version', '1.0.0'), - description=self._manifest.get( - 'description', - 'Notifications and announcements management' - ), - 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/notifications'), - ) - - def get_blueprint(self) -> Optional[Blueprint]: - """Return Flask Blueprint with API routes.""" - return notifications_bp - - def get_models(self) -> List[Type]: - """Return list of SQLAlchemy model classes.""" - return [Notification, NotificationType] - - def init_app(self, app: Flask, db_instance) -> None: - """Initialize plugin with Flask app.""" - logger.info(f"Notifications plugin initialized (v{self.meta.version})") - - def on_install(self, app: Flask) -> None: - """Called when plugin is installed.""" - with app.app_context(): - self._ensure_notification_types() - logger.info("Notifications plugin installed") - - def _ensure_notification_types(self) -> None: - """Ensure default notification types exist.""" - default_types = [ - ('Awareness', 'General awareness notification', '#17a2b8', 'info-circle'), - ('Change', 'Planned change notification', '#ffc107', 'exchange-alt'), - ('Incident', 'Incident or outage notification', '#dc3545', 'exclamation-triangle'), - ('Maintenance', 'Scheduled maintenance notification', '#6c757d', 'wrench'), - ('General', 'General announcement', '#28a745', 'bullhorn'), - ] - - for typename, description, color, icon in default_types: - existing = NotificationType.query.filter_by(typename=typename).first() - if not existing: - t = NotificationType( - typename=typename, - description=description, - color=color, - icon=icon - ) - db.session.add(t) - logger.debug(f"Created notification type: {typename}") - - db.session.commit() - - def on_uninstall(self, app: Flask) -> None: - """Called when plugin is uninstalled.""" - logger.info("Notifications plugin uninstalled") - - def get_cli_commands(self) -> List: - """Return CLI commands for this plugin.""" - - @click.group('notifications') - def notifications_cli(): - """Notifications plugin commands.""" - pass - - @notifications_cli.command('list-types') - def list_types(): - """List all notification types.""" - from flask import current_app - - with current_app.app_context(): - types = NotificationType.query.filter_by(isactive=True).all() - if not types: - click.echo('No notification types found.') - return - - click.echo('Notification Types:') - for t in types: - click.echo(f" [{t.notificationtypeid}] {t.typename} ({t.color})") - - @notifications_cli.command('stats') - def stats(): - """Show notification statistics.""" - from flask import current_app - from datetime import datetime - - with current_app.app_context(): - now = datetime.utcnow() - - total = Notification.query.filter( - Notification.isactive == True - ).count() - - active = Notification.query.filter( - Notification.isactive == True, - Notification.startdate <= now, - db.or_( - Notification.enddate.is_(None), - Notification.enddate >= now - ) - ).count() - - click.echo(f"Total notifications: {total}") - click.echo(f"Currently active: {active}") - - @notifications_cli.command('create') - @click.option('--title', required=True, help='Notification title') - @click.option('--message', required=True, help='Notification message') - @click.option('--type', 'type_name', default='General', help='Notification type') - def create_notification(title, message, type_name): - """Create a new notification.""" - from flask import current_app - - with current_app.app_context(): - ntype = NotificationType.query.filter_by(typename=type_name).first() - if not ntype: - click.echo(f"Error: Notification type '{type_name}' not found.") - return - - n = Notification( - title=title, - message=message, - notificationtypeid=ntype.notificationtypeid - ) - db.session.add(n) - db.session.commit() - - click.echo(f"Created notification #{n.notificationid}: {title}") - - return [notifications_cli] - - def get_dashboard_widgets(self) -> List[Dict]: - """Return dashboard widget definitions.""" - return [ - { - 'name': 'Active Notifications', - 'component': 'NotificationsWidget', - 'endpoint': '/api/notifications/dashboard/summary', - 'size': 'small', - 'position': 1, - }, - ] - - def get_navigation_items(self) -> List[Dict]: - """Return navigation menu items.""" - return [ - { - 'name': 'Notifications', - 'icon': 'bell', - 'route': '/notifications', - 'position': 5, - }, - { - 'name': 'Calendar', - 'icon': 'calendar', - 'route': '/calendar', - 'position': 6, - }, - ] +"""Notifications 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 + +from .models import Notification, NotificationType +from .api import notifications_bp + +logger = logging.getLogger(__name__) + + +class NotificationsPlugin(BasePlugin): + """ + Notifications plugin - manages announcements and notifications. + + Provides functionality for: + - Creating and managing notifications/announcements + - Displaying banner notifications + - Calendar view of notifications + """ + + def __init__(self): + self._manifest = self._load_manifest() + + def _load_manifest(self) -> Dict: + """Load plugin manifest from JSON file.""" + manifest_path = Path(__file__).parent / 'manifest.json' + if manifest_path.exists(): + with open(manifest_path, 'r') as f: + return json.load(f) + return {} + + @property + def meta(self) -> PluginMeta: + """Return plugin metadata.""" + return PluginMeta( + name=self._manifest.get('name', 'notifications'), + version=self._manifest.get('version', '1.0.0'), + description=self._manifest.get( + 'description', + 'Notifications and announcements management' + ), + 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/notifications'), + ) + + def get_blueprint(self) -> Optional[Blueprint]: + """Return Flask Blueprint with API routes.""" + return notifications_bp + + def get_models(self) -> List[Type]: + """Return list of SQLAlchemy model classes.""" + return [Notification, NotificationType] + + def init_app(self, app: Flask, db_instance) -> None: + """Initialize plugin with Flask app.""" + logger.info(f"Notifications plugin initialized (v{self.meta.version})") + + def on_install(self, app: Flask) -> None: + """Called when plugin is installed.""" + with app.app_context(): + self._ensure_notification_types() + logger.info("Notifications plugin installed") + + def _ensure_notification_types(self) -> None: + """Ensure default notification types exist.""" + default_types = [ + ('Awareness', 'General awareness notification', '#17a2b8', 'info-circle'), + ('Change', 'Planned change notification', '#ffc107', 'exchange-alt'), + ('Incident', 'Incident or outage notification', '#dc3545', 'exclamation-triangle'), + ('Maintenance', 'Scheduled maintenance notification', '#6c757d', 'wrench'), + ('General', 'General announcement', '#28a745', 'bullhorn'), + ] + + for typename, description, color, icon in default_types: + existing = NotificationType.query.filter_by(typename=typename).first() + if not existing: + t = NotificationType( + typename=typename, + description=description, + color=color, + icon=icon + ) + db.session.add(t) + logger.debug(f"Created notification type: {typename}") + + db.session.commit() + + def on_uninstall(self, app: Flask) -> None: + """Called when plugin is uninstalled.""" + logger.info("Notifications plugin uninstalled") + + def get_cli_commands(self) -> List: + """Return CLI commands for this plugin.""" + + @click.group('notifications') + def notifications_cli(): + """Notifications plugin commands.""" + pass + + @notifications_cli.command('list-types') + def list_types(): + """List all notification types.""" + from flask import current_app + + with current_app.app_context(): + types = NotificationType.query.filter_by(isactive=True).all() + if not types: + click.echo('No notification types found.') + return + + click.echo('Notification Types:') + for t in types: + click.echo(f" [{t.notificationtypeid}] {t.typename} ({t.color})") + + @notifications_cli.command('stats') + def stats(): + """Show notification statistics.""" + from flask import current_app + from datetime import datetime + + with current_app.app_context(): + now = datetime.utcnow() + + total = Notification.query.filter( + Notification.isactive == True + ).count() + + active = Notification.query.filter( + Notification.isactive == True, + Notification.startdate <= now, + db.or_( + Notification.enddate.is_(None), + Notification.enddate >= now + ) + ).count() + + click.echo(f"Total notifications: {total}") + click.echo(f"Currently active: {active}") + + @notifications_cli.command('create') + @click.option('--title', required=True, help='Notification title') + @click.option('--message', required=True, help='Notification message') + @click.option('--type', 'type_name', default='General', help='Notification type') + def create_notification(title, message, type_name): + """Create a new notification.""" + from flask import current_app + + with current_app.app_context(): + ntype = NotificationType.query.filter_by(typename=type_name).first() + if not ntype: + click.echo(f"Error: Notification type '{type_name}' not found.") + return + + n = Notification( + title=title, + message=message, + notificationtypeid=ntype.notificationtypeid + ) + db.session.add(n) + db.session.commit() + + click.echo(f"Created notification #{n.notificationid}: {title}") + + return [notifications_cli] + + def get_dashboard_widgets(self) -> List[Dict]: + """Return dashboard widget definitions.""" + return [ + { + 'name': 'Active Notifications', + 'component': 'NotificationsWidget', + 'endpoint': '/api/notifications/dashboard/summary', + 'size': 'small', + 'position': 1, + }, + ] + + def get_navigation_items(self) -> List[Dict]: + """Return navigation menu items.""" + return [ + { + 'name': 'Notifications', + 'icon': 'bell', + 'route': '/notifications', + 'position': 5, + }, + { + 'name': 'Calendar', + 'icon': 'calendar', + 'route': '/calendar', + 'position': 6, + }, + ] diff --git a/plugins/printers/api/asset_routes.py b/plugins/printers/api/asset_routes.py index ef3b2fe..d5f7c91 100644 --- a/plugins/printers/api/asset_routes.py +++ b/plugins/printers/api/asset_routes.py @@ -5,15 +5,7 @@ import logging from flask import Blueprint, request from flask_jwt_extended import jwt_required -from shopdb.extensions import db, cache -from shopdb.core.models import Asset, AssetType, Vendor, Model, Communication, CommunicationType -from shopdb.utils.responses import ( - success_response, - error_response, - paginated_response, - ErrorCodes -) -from shopdb.utils.pagination import get_pagination_params, paginate_query +from shopdb.api import db, cache, Asset, AssetType, Vendor, Model, Communication, CommunicationType, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query from ..models import Printer, PrinterType, ModelSupply from ..models.model_supply import SUPPLY_TYPES, SUPPLY_COLORS, CAPACITY_TIERS @@ -565,7 +557,7 @@ def _get_low_supplies_data(): # location name for the report row location_name = None if asset.locationid: - from shopdb.core.models import Location + from shopdb.api import Location loc = Location.query.get(asset.locationid) if loc: location_name = loc.locationname diff --git a/plugins/printers/models/model_supply.py b/plugins/printers/models/model_supply.py index 398cc93..05d7575 100644 --- a/plugins/printers/models/model_supply.py +++ b/plugins/printers/models/model_supply.py @@ -6,8 +6,7 @@ drum/waste/maintenance item). Lets new models and their toners be added through the API/UI without a code change. """ -from shopdb.extensions import db -from shopdb.core.models.base import BaseModel +from shopdb.api import db, BaseModel # allowed values, surfaced to the UI via the /supplies/meta endpoint diff --git a/plugins/printers/models/printer.py b/plugins/printers/models/printer.py index 8698e9c..beab8d8 100644 --- a/plugins/printers/models/printer.py +++ b/plugins/printers/models/printer.py @@ -1,122 +1,121 @@ -"""Printer plugin models - new Asset-based architecture.""" - -from shopdb.extensions import db -from shopdb.core.models.base import BaseModel - - -class PrinterType(BaseModel): - """ - Printer type classification. - - Examples: Laser, Inkjet, Label, MFP, Plotter, etc. - """ - __tablename__ = 'printertypes' - - printertypeid = db.Column(db.Integer, primary_key=True) - printertype = db.Column(db.String(100), unique=True, nullable=False) - description = db.Column(db.Text) - icon = db.Column(db.String(50), comment='Icon name for UI') - - def __repr__(self): - return f"" - - -class Printer(BaseModel): - """ - Printer-specific extension data (new Asset architecture). - - Links to core Asset table via assetid. - Stores printer-specific fields like type, Windows name, share name, etc. - """ - __tablename__ = 'printers' - - printerid = db.Column(db.Integer, primary_key=True) - - # Link to core asset - assetid = db.Column( - db.Integer, - db.ForeignKey('assets.assetid', ondelete='CASCADE'), - unique=True, - nullable=False, - index=True - ) - - # Printer classification - printertypeid = db.Column( - db.Integer, - db.ForeignKey('printertypes.printertypeid'), - nullable=True - ) - - # Vendor - vendorid = db.Column( - db.Integer, - db.ForeignKey('vendors.vendorid'), - nullable=True - ) - modelnumberid = db.Column( - db.Integer, - db.ForeignKey('models.modelnumberid'), - nullable=True - ) - - # Network identity - hostname = db.Column( - db.String(100), - index=True, - comment='Network hostname' - ) - - # Windows/Network naming - windowsname = db.Column( - db.String(255), - comment='Windows printer name (e.g., \\\\server\\printer)' - ) - sharename = db.Column( - db.String(100), - comment='CSF/share name' - ) - - # Installation - iscsf = db.Column(db.Boolean, default=False, comment='Is CSF printer') - installpath = db.Column(db.String(255), comment='Driver install path') - - # Printer PIN (for secure print) - pin = db.Column(db.String(20)) - - # Features - iscolor = db.Column(db.Boolean, default=False, comment='Color capable') - isduplex = db.Column(db.Boolean, default=False, comment='Duplex capable') - isnetwork = db.Column(db.Boolean, default=True, comment='Network connected') - - # Relationships - asset = db.relationship( - 'Asset', - backref=db.backref('printer', uselist=False, lazy='joined') - ) - printertype = db.relationship('PrinterType', backref='printers') - vendor = db.relationship('Vendor', backref='printer_items') - model = db.relationship('Model', backref='printer_items') - - __table_args__ = ( - db.Index('idx_printer_type', 'printertypeid'), - db.Index('idx_printer_hostname', 'hostname'), - db.Index('idx_printer_windowsname', 'windowsname'), - ) - - def __repr__(self): - return f"" - - def to_dict(self): - """Convert to dictionary with related names.""" - result = super().to_dict() - - # Add related object names - if self.printertype: - result['printertypename'] = self.printertype.printertype - if self.vendor: - result['vendorname'] = self.vendor.vendor - if self.model: - result['modelname'] = self.model.modelnumber - - return result +"""Printer plugin models - new Asset-based architecture.""" + +from shopdb.api import db, BaseModel + + +class PrinterType(BaseModel): + """ + Printer type classification. + + Examples: Laser, Inkjet, Label, MFP, Plotter, etc. + """ + __tablename__ = 'printertypes' + + printertypeid = db.Column(db.Integer, primary_key=True) + printertype = db.Column(db.String(100), unique=True, nullable=False) + description = db.Column(db.Text) + icon = db.Column(db.String(50), comment='Icon name for UI') + + def __repr__(self): + return f"" + + +class Printer(BaseModel): + """ + Printer-specific extension data (new Asset architecture). + + Links to core Asset table via assetid. + Stores printer-specific fields like type, Windows name, share name, etc. + """ + __tablename__ = 'printers' + + printerid = db.Column(db.Integer, primary_key=True) + + # Link to core asset + assetid = db.Column( + db.Integer, + db.ForeignKey('assets.assetid', ondelete='CASCADE'), + unique=True, + nullable=False, + index=True + ) + + # Printer classification + printertypeid = db.Column( + db.Integer, + db.ForeignKey('printertypes.printertypeid'), + nullable=True + ) + + # Vendor + vendorid = db.Column( + db.Integer, + db.ForeignKey('vendors.vendorid'), + nullable=True + ) + modelnumberid = db.Column( + db.Integer, + db.ForeignKey('models.modelnumberid'), + nullable=True + ) + + # Network identity + hostname = db.Column( + db.String(100), + index=True, + comment='Network hostname' + ) + + # Windows/Network naming + windowsname = db.Column( + db.String(255), + comment='Windows printer name (e.g., \\\\server\\printer)' + ) + sharename = db.Column( + db.String(100), + comment='CSF/share name' + ) + + # Installation + iscsf = db.Column(db.Boolean, default=False, comment='Is CSF printer') + installpath = db.Column(db.String(255), comment='Driver install path') + + # Printer PIN (for secure print) + pin = db.Column(db.String(20)) + + # Features + iscolor = db.Column(db.Boolean, default=False, comment='Color capable') + isduplex = db.Column(db.Boolean, default=False, comment='Duplex capable') + isnetwork = db.Column(db.Boolean, default=True, comment='Network connected') + + # Relationships + asset = db.relationship( + 'Asset', + backref=db.backref('printer', uselist=False, lazy='joined') + ) + printertype = db.relationship('PrinterType', backref='printers') + vendor = db.relationship('Vendor', backref='printer_items') + model = db.relationship('Model', backref='printer_items') + + __table_args__ = ( + db.Index('idx_printer_type', 'printertypeid'), + db.Index('idx_printer_hostname', 'hostname'), + db.Index('idx_printer_windowsname', 'windowsname'), + ) + + def __repr__(self): + return f"" + + def to_dict(self): + """Convert to dictionary with related names.""" + result = super().to_dict() + + # Add related object names + if self.printertype: + result['printertypename'] = self.printertype.printertype + if self.vendor: + result['vendorname'] = self.vendor.vendor + if self.model: + result['modelname'] = self.model.modelnumber + + return result diff --git a/plugins/printers/plugin.py b/plugins/printers/plugin.py index ac59f96..d647497 100644 --- a/plugins/printers/plugin.py +++ b/plugins/printers/plugin.py @@ -9,9 +9,7 @@ from flask import Flask, Blueprint import click from shopdb.plugins.base import BasePlugin, PluginMeta -from shopdb.extensions import db -from shopdb.core.models.machine import MachineType -from shopdb.core.models import AssetType +from shopdb.api import db, AssetType from .models import Printer, PrinterType, ModelSupply from .api import printers_asset_bp @@ -104,7 +102,6 @@ class PrintersPlugin(BasePlugin): with app.app_context(): self._ensure_asset_type() self._ensure_printer_types() - self._ensure_legacy_machine_types() logger.info("Printers plugin installed") def _ensure_asset_type(self) -> None: @@ -149,30 +146,6 @@ class PrintersPlugin(BasePlugin): db.session.commit() - def _ensure_legacy_machine_types(self) -> None: - """Ensure basic printer machine types exist (legacy architecture).""" - printertypes = [ - ('Laser Printer', 'Printer', 'Standard laser printer'), - ('Inkjet Printer', 'Printer', 'Inkjet printer'), - ('Label Printer', 'Printer', 'Label/barcode printer'), - ('Multifunction Printer', 'Printer', 'MFP with scan/copy/fax'), - ('Plotter', 'Printer', 'Large format plotter'), - ] - - for name, category, description in printertypes: - existing = MachineType.query.filter_by(machinetype=name).first() - if not existing: - mt = MachineType( - machinetype=name, - category=category, - description=description, - icon='printer' - ) - db.session.add(mt) - logger.debug(f"Created machine type: {name}") - - db.session.commit() - def on_uninstall(self, app: Flask) -> None: """Called when plugin is uninstalled.""" logger.info("Printers plugin uninstalled") diff --git a/plugins/printers/services/seed_supplies.py b/plugins/printers/services/seed_supplies.py index 9bfcb67..82e69b7 100644 --- a/plugins/printers/services/seed_supplies.py +++ b/plugins/printers/services/seed_supplies.py @@ -19,9 +19,7 @@ Key facts encoded here: import logging -from shopdb.extensions import db -from shopdb.core.models import Vendor, Model -from shopdb.core.models.machine import MachineType +from shopdb.api import db, Vendor, Model from ..models import ModelSupply @@ -311,9 +309,6 @@ def seedsupplies(): model that matches a family's keys; if a family matches no existing model, creates a canonical model row so its toners are still available. """ - printertype = MachineType.query.filter_by(category='Printer').first() - printertypeid = printertype.machinetypeid if printertype else None - models_touched = 0 supplies_added = 0 @@ -322,10 +317,11 @@ def seedsupplies(): targets = _matching_models(family['matchkeys'], vendor.vendorid) if not targets: + # machinetypeid is a legacy Model column (nullable); printers are + # asset-based now and carry their type via PrinterType, not here. model = Model( modelnumber=family['canonical'], vendorid=vendor.vendorid, - machinetypeid=printertypeid, ) db.session.add(model) db.session.flush() diff --git a/plugins/printers/services/zabbix_service.py b/plugins/printers/services/zabbix_service.py index d2465bb..ddd8e60 100644 --- a/plugins/printers/services/zabbix_service.py +++ b/plugins/printers/services/zabbix_service.py @@ -24,7 +24,7 @@ from typing import Dict, List, Optional import requests from flask import current_app -from shopdb.extensions import cache +from shopdb.api import cache logger = logging.getLogger(__name__) @@ -55,7 +55,7 @@ class ZabbixService: @property def isenabled(self) -> bool: """Whether the integration is switched on.""" - from shopdb.core.models import Setting + from shopdb.api import Setting db_enabled = Setting.get('zabbix_enabled') if db_enabled is not None: return bool(db_enabled) @@ -66,7 +66,7 @@ class ZabbixService: """Enabled, and a URL plus token are present.""" if not self.isenabled: return False - from shopdb.core.models import Setting + from shopdb.api import Setting self._url = Setting.get('zabbix_url') or current_app.config.get('ZABBIX_URL') self._token = Setting.get('zabbix_token') or current_app.config.get('ZABBIX_TOKEN') return bool(self._url and self._token) diff --git a/plugins/usb/api/routes.py b/plugins/usb/api/routes.py index a990ca5..d17a664 100644 --- a/plugins/usb/api/routes.py +++ b/plugins/usb/api/routes.py @@ -4,15 +4,7 @@ from flask import Blueprint, request from flask_jwt_extended import jwt_required, get_jwt_identity from datetime import datetime -from shopdb.extensions import db -from shopdb.core.models import AuditLog -from shopdb.utils.responses import ( - success_response, - error_response, - paginated_response, - ErrorCodes -) -from shopdb.utils.pagination import get_pagination_params, paginate_query +from shopdb.api import db, AuditLog, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query from ..models import USBDevice, USBDeviceType, USBCheckout diff --git a/plugins/usb/models/usb_device.py b/plugins/usb/models/usb_device.py index ab43681..f37490b 100644 --- a/plugins/usb/models/usb_device.py +++ b/plugins/usb/models/usb_device.py @@ -1,167 +1,166 @@ -"""USB device plugin models.""" - -from datetime import datetime -from shopdb.extensions import db -from shopdb.core.models.base import BaseModel, AuditMixin - - -class USBDeviceType(BaseModel): - """ - USB device type classification. - - Examples: Flash Drive, External HDD, External SSD, Card Reader - """ - __tablename__ = 'usbdevicetypes' - - usbdevicetypeid = db.Column(db.Integer, primary_key=True) - typename = db.Column(db.String(50), unique=True, nullable=False) - description = db.Column(db.Text) - icon = db.Column(db.String(50), default='usb', comment='Icon name for UI') - - def __repr__(self): - return f"" - - -class USBDevice(BaseModel, AuditMixin): - """ - USB device model. - - Tracks USB storage devices that can be checked out by users. - """ - __tablename__ = 'usbdevices' - - usbdeviceid = db.Column(db.Integer, primary_key=True) - - # Identification - serialnumber = db.Column(db.String(100), unique=True, nullable=False) - label = db.Column(db.String(100), nullable=True, comment='Human-readable label') - assetnumber = db.Column(db.String(50), nullable=True, comment='Optional asset tag') - - # Classification - usbdevicetypeid = db.Column( - db.Integer, - db.ForeignKey('usbdevicetypes.usbdevicetypeid'), - nullable=True - ) - - # Specifications - capacitygb = db.Column(db.Integer, nullable=True, comment='Capacity in GB') - vendorid = db.Column(db.String(10), nullable=True, comment='USB Vendor ID (hex)') - productid = db.Column(db.String(10), nullable=True, comment='USB Product ID (hex)') - manufacturer = db.Column(db.String(100), nullable=True) - productname = db.Column(db.String(100), nullable=True) - - # Current status - ischeckedout = db.Column(db.Boolean, default=False) - currentuserid = db.Column(db.String(50), nullable=True, comment='SSO of current user') - currentusername = db.Column(db.String(100), nullable=True, comment='Name of current user') - currentcheckoutdate = db.Column(db.DateTime, nullable=True) - - # Location - storagelocation = db.Column(db.String(200), nullable=True, comment='Where device is stored when not checked out') - - # Security - pin = db.Column(db.String(50), nullable=True, comment='PIN for encrypted devices') - - # Notes - notes = db.Column(db.Text, nullable=True) - - # Relationships - devicetype = db.relationship('USBDeviceType', backref='devices') - - # Indexes - __table_args__ = ( - db.Index('idx_usb_serial', 'serialnumber'), - db.Index('idx_usb_checkedout', 'ischeckedout'), - db.Index('idx_usb_type', 'usbdevicetypeid'), - db.Index('idx_usb_currentuser', 'currentuserid'), - ) - - def __repr__(self): - return f"" - - @property - def display_name(self): - """Get display name (label if set, otherwise serial number).""" - return self.label or self.serialnumber - - def to_dict(self): - """Convert to dictionary with related data.""" - result = super().to_dict() - - # Add type info - if self.devicetype: - result['typename'] = self.devicetype.typename - result['typeicon'] = self.devicetype.icon - - # Add computed property - result['displayname'] = self.display_name - - return result - - -class USBCheckout(BaseModel): - """ - USB device checkout history. - - Tracks when devices are checked out and returned. - Maps to existing usbcheckouts table from classic ShopDB. - """ - __tablename__ = 'usbcheckouts' - - checkoutid = db.Column(db.Integer, primary_key=True) - - # Device reference (new column linking to usbdevices table) - usbdeviceid = db.Column( - db.Integer, - db.ForeignKey('usbdevices.usbdeviceid', ondelete='CASCADE'), - nullable=True - ) - - # Legacy reference to machines table (kept for backward compatibility) - machineid = db.Column(db.Integer, nullable=False) - - # User info - sso = db.Column(db.String(20), nullable=False, comment='SSO of user') - checkoutname = db.Column(db.String(100), nullable=True, comment='Name of user') - - # Checkout details - checkouttime = db.Column(db.DateTime, nullable=False, default=datetime.utcnow) - checkintime = db.Column(db.DateTime, nullable=True) - - # Metadata - checkoutreason = db.Column(db.Text, nullable=True, comment='Reason for checkout') - checkinnotes = db.Column(db.Text, nullable=True) - waswiped = db.Column(db.Boolean, nullable=True, comment='Was device wiped after return') - - # Relationships - device = db.relationship('USBDevice', backref=db.backref('checkouts', lazy='dynamic')) - - def __repr__(self): - return f"" - - @property - def is_active(self): - """Check if this checkout is currently active (not returned).""" - return self.checkintime is None - - @property - def duration_days(self): - """Get duration of checkout in days.""" - end = self.checkintime or datetime.utcnow() - delta = end - self.checkouttime - return delta.days - - def to_dict(self): - """Convert to dictionary with computed fields.""" - result = super().to_dict() - - result['isactivecheckout'] = self.is_active - result['durationdays'] = self.duration_days - - # Add device info if loaded - if self.device: - result['devicelabel'] = self.device.label - result['deviceserialnumber'] = self.device.serialnumber - - return result +"""USB device plugin models.""" + +from datetime import datetime +from shopdb.api import db, BaseModel, AuditMixin + + +class USBDeviceType(BaseModel): + """ + USB device type classification. + + Examples: Flash Drive, External HDD, External SSD, Card Reader + """ + __tablename__ = 'usbdevicetypes' + + usbdevicetypeid = db.Column(db.Integer, primary_key=True) + typename = db.Column(db.String(50), unique=True, nullable=False) + description = db.Column(db.Text) + icon = db.Column(db.String(50), default='usb', comment='Icon name for UI') + + def __repr__(self): + return f"" + + +class USBDevice(BaseModel, AuditMixin): + """ + USB device model. + + Tracks USB storage devices that can be checked out by users. + """ + __tablename__ = 'usbdevices' + + usbdeviceid = db.Column(db.Integer, primary_key=True) + + # Identification + serialnumber = db.Column(db.String(100), unique=True, nullable=False) + label = db.Column(db.String(100), nullable=True, comment='Human-readable label') + assetnumber = db.Column(db.String(50), nullable=True, comment='Optional asset tag') + + # Classification + usbdevicetypeid = db.Column( + db.Integer, + db.ForeignKey('usbdevicetypes.usbdevicetypeid'), + nullable=True + ) + + # Specifications + capacitygb = db.Column(db.Integer, nullable=True, comment='Capacity in GB') + vendorid = db.Column(db.String(10), nullable=True, comment='USB Vendor ID (hex)') + productid = db.Column(db.String(10), nullable=True, comment='USB Product ID (hex)') + manufacturer = db.Column(db.String(100), nullable=True) + productname = db.Column(db.String(100), nullable=True) + + # Current status + ischeckedout = db.Column(db.Boolean, default=False) + currentuserid = db.Column(db.String(50), nullable=True, comment='SSO of current user') + currentusername = db.Column(db.String(100), nullable=True, comment='Name of current user') + currentcheckoutdate = db.Column(db.DateTime, nullable=True) + + # Location + storagelocation = db.Column(db.String(200), nullable=True, comment='Where device is stored when not checked out') + + # Security + pin = db.Column(db.String(50), nullable=True, comment='PIN for encrypted devices') + + # Notes + notes = db.Column(db.Text, nullable=True) + + # Relationships + devicetype = db.relationship('USBDeviceType', backref='devices') + + # Indexes + __table_args__ = ( + db.Index('idx_usb_serial', 'serialnumber'), + db.Index('idx_usb_checkedout', 'ischeckedout'), + db.Index('idx_usb_type', 'usbdevicetypeid'), + db.Index('idx_usb_currentuser', 'currentuserid'), + ) + + def __repr__(self): + return f"" + + @property + def display_name(self): + """Get display name (label if set, otherwise serial number).""" + return self.label or self.serialnumber + + def to_dict(self): + """Convert to dictionary with related data.""" + result = super().to_dict() + + # Add type info + if self.devicetype: + result['typename'] = self.devicetype.typename + result['typeicon'] = self.devicetype.icon + + # Add computed property + result['displayname'] = self.display_name + + return result + + +class USBCheckout(BaseModel): + """ + USB device checkout history. + + Tracks when devices are checked out and returned. + Maps to existing usbcheckouts table from classic ShopDB. + """ + __tablename__ = 'usbcheckouts' + + checkoutid = db.Column(db.Integer, primary_key=True) + + # Device reference (new column linking to usbdevices table) + usbdeviceid = db.Column( + db.Integer, + db.ForeignKey('usbdevices.usbdeviceid', ondelete='CASCADE'), + nullable=True + ) + + # Legacy reference to machines table (kept for backward compatibility) + machineid = db.Column(db.Integer, nullable=False) + + # User info + sso = db.Column(db.String(20), nullable=False, comment='SSO of user') + checkoutname = db.Column(db.String(100), nullable=True, comment='Name of user') + + # Checkout details + checkouttime = db.Column(db.DateTime, nullable=False, default=datetime.utcnow) + checkintime = db.Column(db.DateTime, nullable=True) + + # Metadata + checkoutreason = db.Column(db.Text, nullable=True, comment='Reason for checkout') + checkinnotes = db.Column(db.Text, nullable=True) + waswiped = db.Column(db.Boolean, nullable=True, comment='Was device wiped after return') + + # Relationships + device = db.relationship('USBDevice', backref=db.backref('checkouts', lazy='dynamic')) + + def __repr__(self): + return f"" + + @property + def is_active(self): + """Check if this checkout is currently active (not returned).""" + return self.checkintime is None + + @property + def duration_days(self): + """Get duration of checkout in days.""" + end = self.checkintime or datetime.utcnow() + delta = end - self.checkouttime + return delta.days + + def to_dict(self): + """Convert to dictionary with computed fields.""" + result = super().to_dict() + + result['isactivecheckout'] = self.is_active + result['durationdays'] = self.duration_days + + # Add device info if loaded + if self.device: + result['devicelabel'] = self.device.label + result['deviceserialnumber'] = self.device.serialnumber + + return result diff --git a/plugins/usb/plugin.py b/plugins/usb/plugin.py index 0952a71..a1ed234 100644 --- a/plugins/usb/plugin.py +++ b/plugins/usb/plugin.py @@ -8,7 +8,7 @@ from typing import List, Dict, Optional, Type from flask import Flask, Blueprint from shopdb.plugins.base import BasePlugin, PluginMeta -from shopdb.extensions import db +from shopdb.api import db from .models import USBDevice, USBDeviceType, USBCheckout from .api import usb_bp diff --git a/shopdb/__init__.py b/shopdb/__init__.py index 7eab542..cfd7ee8 100644 --- a/shopdb/__init__.py +++ b/shopdb/__init__.py @@ -12,7 +12,10 @@ from .plugins import plugin_manager # ADR-002 for the bump rules. Plugins declare a compatible range in # their manifest.json `core_version` field. Pre-1.0 (0.x) means the # contract is still settling; sister sites should pin tight ranges. -__contract_version__ = '0.2.0' +# 0.3.0: shopdb.api expanded to the full plugin import surface (db, cache, +# model bases, core models, response + pagination helpers, employee_connection) +# so plugins no longer import internal core paths. Additive, hence minor bump. +__contract_version__ = '0.3.0' def create_app(config_name: str = None) -> Flask: diff --git a/shopdb/api/__init__.py b/shopdb/api/__init__.py index 6439f09..0a03eb9 100644 --- a/shopdb/api/__init__.py +++ b/shopdb/api/__init__.py @@ -14,7 +14,47 @@ Setting helpers are exposed via BasePlugin instance methods from typing import Any, Dict, Optional -from shopdb.core.models import AuditLog +# -- Plugin contract surface (ADR-001, versioned per ADR-002) ---------------- +# Everything a plugin is allowed to import from the core lives here. Plugins +# import these from `shopdb.api`, never from internal paths like +# `shopdb.core.models.*` or `shopdb.extensions`. The contract test +# (tests/test_plugin_contract.py) enforces this. Adding a name here is an +# additive (minor) contract change; removing one is breaking (major). + +# Infrastructure +from shopdb.extensions import db, cache + +# Model base classes for declaring plugin tables +from shopdb.core.models.base import BaseModel, AuditMixin + +# Core domain models plugins legitimately reference (the asset contract) +from shopdb.core.models import ( + Asset, + AssetType, + AssetStatus, + Vendor, + Model, + Communication, + CommunicationType, + Location, + Setting, + AuditLog, + Application, + AppVersion, + OperatingSystem, +) + +# Response + pagination helpers for plugin API blueprints +from shopdb.utils.responses import ( + success_response, + error_response, + paginated_response, + ErrorCodes, +) +from shopdb.utils.pagination import get_pagination_params, paginate_query + +# Legacy employee directory lookup (read-only) used by notifications +from shopdb.utils.employee_db import employee_connection def audit_log( @@ -155,4 +195,37 @@ def resolve_asset_position(asset) -> Optional[Dict[str, Any]]: return None -__all__ = ['audit_log', 'resolve_asset_position'] +__all__ = [ + # Helpers + 'audit_log', + 'resolve_asset_position', + # Infrastructure + 'db', + 'cache', + # Model bases + 'BaseModel', + 'AuditMixin', + # Core models + 'Asset', + 'AssetType', + 'AssetStatus', + 'Vendor', + 'Model', + 'Communication', + 'CommunicationType', + 'Location', + 'Setting', + 'AuditLog', + 'Application', + 'AppVersion', + 'OperatingSystem', + # Response + pagination helpers + 'success_response', + 'error_response', + 'paginated_response', + 'ErrorCodes', + 'get_pagination_params', + 'paginate_query', + # Legacy employee directory + 'employee_connection', +] diff --git a/shopdb/plugins/templates/api/routes.py.tmpl b/shopdb/plugins/templates/api/routes.py.tmpl index bfe0d28..7460022 100644 --- a/shopdb/plugins/templates/api/routes.py.tmpl +++ b/shopdb/plugins/templates/api/routes.py.tmpl @@ -3,13 +3,14 @@ from flask import Blueprint, request from flask_jwt_extended import jwt_required -from shopdb.utils.responses import ( +from shopdb.api import ( success_response, error_response, paginated_response, ErrorCodes, + get_pagination_params, + paginate_query, ) -from shopdb.utils.pagination import get_pagination_params, paginate_query from ..models import $Name diff --git a/shopdb/plugins/templates/models/model.py.tmpl b/shopdb/plugins/templates/models/model.py.tmpl index 5d3d42f..5669baf 100644 --- a/shopdb/plugins/templates/models/model.py.tmpl +++ b/shopdb/plugins/templates/models/model.py.tmpl @@ -6,8 +6,7 @@ this table holds the $name-specific fields. Replace the example fields below with your domain model. """ -from shopdb.extensions import db -from shopdb.core.models.base import BaseModel +from shopdb.api import db, BaseModel class $Name(BaseModel): diff --git a/shopdb/plugins/templates/plugin.py.tmpl b/shopdb/plugins/templates/plugin.py.tmpl index ceb474d..2daf77c 100644 --- a/shopdb/plugins/templates/plugin.py.tmpl +++ b/shopdb/plugins/templates/plugin.py.tmpl @@ -11,8 +11,7 @@ from typing import List, Dict, Optional, Type from flask import Flask, Blueprint from shopdb.plugins.base import BasePlugin, PluginMeta -from shopdb.core.models import AssetType -from shopdb.extensions import db +from shopdb.api import db, AssetType from .models import $Name from .api import ${name}_bp diff --git a/tests/test_plugin_contract.py b/tests/test_plugin_contract.py index a4aa91c..9d9a031 100644 --- a/tests/test_plugin_contract.py +++ b/tests/test_plugin_contract.py @@ -6,6 +6,7 @@ plugin's plugin.py / manifest.json. """ import json +import re from pathlib import Path import pytest @@ -143,3 +144,38 @@ def test_baseplugin_does_not_have_event_handlers_hook(): def test_baseplugin_has_collector_schema_hook(): """The collector schema hook is on the contract surface.""" assert hasattr(BasePlugin, 'get_collector_schema') + + +# Imports a plugin may make from the core. shopdb.api is the contract surface; +# shopdb.plugins.base is the plugin ABC. Anything else (shopdb.core.*, +# shopdb.extensions, shopdb.utils.*) is a contract violation per ADR-001. +ALLOWED_CORE_IMPORTS = ('shopdb.api', 'shopdb.plugins.base') + +_PLUGIN_IMPORT_RE = re.compile( + r'^\s*(?:from (shopdb[\w.]*) import|import (shopdb[\w.]*))', re.MULTILINE +) + + +def _plugin_source_files(): + root = Path(__file__).resolve().parent.parent / 'plugins' + return [p for p in root.rglob('*.py') if '__pycache__' not in p.parts] + + +def test_plugins_only_import_contract_surface(): + """Plugins must import core code only via shopdb.api / shopdb.plugins.base.""" + violations = [] + for path in _plugin_source_files(): + text = path.read_text() + for match in _PLUGIN_IMPORT_RE.finditer(text): + module = match.group(1) or match.group(2) + if not module.startswith('shopdb'): + continue + if any(module == a or module.startswith(a + '.') + for a in ALLOWED_CORE_IMPORTS): + continue + line = text[:match.start()].count('\n') + 1 + violations.append(f'{path.name}:{line} imports {module}') + assert not violations, ( + 'Plugins must import core only via shopdb.api or shopdb.plugins.base. ' + 'Violations:\n' + '\n'.join(violations) + )