ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Migrate frontend to plugin-based asset architecture

Browse Source 
- Add equipmentApi and computersApi to replace legacy machinesApi
- Add controller vendor/model fields to Equipment model and forms
- Fix map marker navigation to use plugin-specific IDs (equipmentid,
  computerid, printerid, networkdeviceid) instead of assetid
- Fix search to use unified Asset table with correct plugin IDs
- Remove legacy printer search that used non-existent field names
- Enable optional JWT auth for detail endpoints (public read access)
- Clean up USB plugin models (remove unused checkout model)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-01-29 16:07:41 -05:00
parent 9c220a4194
commit c3ce69da12
28 changed files with 4123 additions and 3454 deletions
Download Patch File Download Diff File Expand all files Collapse all files

423
plugins/usb/api/routes.py
View File

@@ -1,11 +1,10 @@
"""USB plugin API endpoints."""
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from flask_jwt_extended import jwt_required, get_jwt_identity
from datetime import datetime
from shopdb.extensions import db
from shopdb.core.models import Machine, MachineType, Vendor, Model
from shopdb.utils.responses import (
success_response,
error_response,
@@ -14,171 +13,267 @@ from shopdb.utils.responses import (
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
from ..models import USBCheckout
from ..models import USBDevice, USBDeviceType, USBCheckout
usb_bp = Blueprint('usb', __name__)
def get_usb_machinetype_id():
"""Get the USB Device machine type ID dynamically."""
usb_type = MachineType.query.filter(
MachineType.machinetype.ilike('%usb%')
).first()
return usb_type.machinetypeid if usb_type else None
# =============================================================================
# USB Device Types
# =============================================================================
@usb_bp.route('/types', methods=['GET'])
@jwt_required(optional=True)
def list_device_types():
"""List all USB device types."""
types = USBDeviceType.query.filter_by(isactive=True).order_by(USBDeviceType.typename).all()
return success_response([{
'usbdevicetypeid': t.usbdevicetypeid,
'typename': t.typename,
'description': t.description,
'icon': t.icon
} for t in types])
@usb_bp.route('/types', methods=['POST'])
@jwt_required()
def create_device_type():
"""Create a new USB device type."""
data = request.get_json() or {}
if not data.get('typename'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'typename is required')
if USBDeviceType.query.filter_by(typename=data['typename']).first():
return error_response(ErrorCodes.CONFLICT, 'Type name already exists', http_code=409)
device_type = USBDeviceType(
typename=data['typename'],
description=data.get('description'),
icon=data.get('icon', 'usb')
)
db.session.add(device_type)
db.session.commit()
return success_response({
'usbdevicetypeid': device_type.usbdevicetypeid,
'typename': device_type.typename
}, message='Device type created', http_code=201)
# =============================================================================
# USB Devices
# =============================================================================
@usb_bp.route('', methods=['GET'])
@jwt_required()
@jwt_required(optional=True)
def list_usb_devices():
"""
List all USB devices with checkout status.
Query parameters:
- page, per_page: Pagination
- search: Search by serial number or alias
- search: Search by serial number, label, or asset number
- available: Filter to only available (not checked out) devices
- typeid: Filter by device type ID
"""
page, per_page = get_pagination_params(request)
usb_type_id = get_usb_machinetype_id()
if not usb_type_id:
return success_response([]) # No USB type found
query = USBDevice.query.filter_by(isactive=True)
# Filter by type
if type_id := request.args.get('typeid'):
query = query.filter_by(usbdevicetypeid=int(type_id))
# Filter by checkout status
if request.args.get('available', '').lower() == 'true':
query = query.filter_by(ischeckedout=False)
elif request.args.get('checkedout', '').lower() == 'true':
query = query.filter_by(ischeckedout=True)
# Get USB devices from machines table
query = db.session.query(Machine).filter(
Machine.machinetypeid == usb_type_id,
Machine.isactive == True
)
# Search filter
if search := request.args.get('search'):
query = query.filter(
db.or_(
Machine.serialnumber.ilike(f'%{search}%'),
Machine.alias.ilike(f'%{search}%'),
Machine.machinenumber.ilike(f'%{search}%')
USBDevice.serialnumber.ilike(f'%{search}%'),
USBDevice.label.ilike(f'%{search}%'),
USBDevice.assetnumber.ilike(f'%{search}%'),
USBDevice.manufacturer.ilike(f'%{search}%')
)
)
query = query.order_by(Machine.alias)
query = query.order_by(USBDevice.label, USBDevice.serialnumber)
items, total = paginate_query(query, page, per_page)
# Build response with checkout status
data = []
for device in items:
# Check if currently checked out
active_checkout = USBCheckout.query.filter_by(
machineid=device.machineid,
checkin_time=None
).first()
item = {
'machineid': device.machineid,
'machinenumber': device.machinenumber,
'alias': device.alias,
'serialnumber': device.serialnumber,
'notes': device.notes,
'vendor_name': device.vendor.vendorname if device.vendor else None,
'model_name': device.model.modelnumber if device.model else None,
'is_checked_out': active_checkout is not None,
'current_checkout': active_checkout.to_dict() if active_checkout else None
}
data.append(item)
# Filter by availability if requested
if request.args.get('available', '').lower() == 'true':
data = [d for d in data if not d['is_checked_out']]
total = len(data)
data = [device.to_dict() for device in items]
return paginated_response(data, page, per_page, total)
@usb_bp.route('/<int:device_id>', methods=['GET'])
@usb_bp.route('', methods=['POST'])
@jwt_required()
def create_usb_device():
"""Create a new USB device."""
data = request.get_json() or {}
if not data.get('serialnumber'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'serialnumber is required')
if USBDevice.query.filter_by(serialnumber=data['serialnumber']).first():
return error_response(ErrorCodes.CONFLICT, 'Serial number already exists', http_code=409)
device = USBDevice(
serialnumber=data['serialnumber'],
label=data.get('label'),
assetnumber=data.get('assetnumber'),
usbdevicetypeid=data.get('usbdevicetypeid'),
capacitygb=data.get('capacitygb'),
vendorid=data.get('vendorid'),
productid=data.get('productid'),
manufacturer=data.get('manufacturer'),
productname=data.get('productname'),
storagelocation=data.get('storagelocation'),
pin=data.get('pin'),
notes=data.get('notes'),
ischeckedout=False
)
db.session.add(device)
db.session.commit()
return success_response(device.to_dict(), message='Device created', http_code=201)
@usb_bp.route('/<int:device_id>', methods=['GET'])
@jwt_required(optional=True)
def get_usb_device(device_id: int):
"""Get a single USB device with checkout history."""
device = Machine.query.filter_by(
machineid=device_id,
machinetypeid=get_usb_machinetype_id()
).first()
device = USBDevice.query.filter_by(usbdeviceid=device_id, isactive=True).first()
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device with ID {device_id} not found',
http_code=404
)
# Get checkout history
# Get recent checkout history
checkouts = USBCheckout.query.filter_by(
machineid=device_id
).order_by(USBCheckout.checkout_time.desc()).limit(50).all()
# Check current checkout
active_checkout = next((c for c in checkouts if c.checkin_time is None), None)
result = {
'machineid': device.machineid,
'machinenumber': device.machinenumber,
'alias': device.alias,
'serialnumber': device.serialnumber,
'notes': device.notes,
'vendor_name': device.vendor.vendorname if device.vendor else None,
'model_name': device.model.modelnumber if device.model else None,
'is_checked_out': active_checkout is not None,
'current_checkout': active_checkout.to_dict() if active_checkout else None,
'checkout_history': [c.to_dict() for c in checkouts]
}
usbdeviceid=device_id
).order_by(USBCheckout.checkout_time.desc()).limit(20).all()
result = device.to_dict()
result['checkout_history'] = [c.to_dict() for c in checkouts]
return success_response(result)
@usb_bp.route('/<int:device_id>', methods=['PUT'])
@jwt_required()
def update_usb_device(device_id: int):
"""Update a USB device."""
device = USBDevice.query.filter_by(usbdeviceid=device_id, isactive=True).first()
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device with ID {device_id} not found',
http_code=404
)
data = request.get_json() or {}
# Update allowed fields
for field in ['label', 'assetnumber', 'usbdevicetypeid', 'capacitygb',
'vendorid', 'productid', 'manufacturer', 'productname',
'storagelocation', 'pin', 'notes']:
if field in data:
setattr(device, field, data[field])
device.modifieddate = datetime.utcnow()
db.session.commit()
return success_response(device.to_dict(), message='Device updated')
@usb_bp.route('/<int:device_id>', methods=['DELETE'])
@jwt_required()
def delete_usb_device(device_id: int):
"""Soft delete a USB device."""
device = USBDevice.query.filter_by(usbdeviceid=device_id, isactive=True).first()
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device with ID {device_id} not found',
http_code=404
)
if device.ischeckedout:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'Cannot delete a device that is currently checked out',
http_code=400
)
device.isactive = False
device.modifieddate = datetime.utcnow()
db.session.commit()
return success_response(None, message='Device deleted')
# =============================================================================
# Checkout/Checkin Operations
# =============================================================================
@usb_bp.route('/<int:device_id>/checkout', methods=['POST'])
@jwt_required()
def checkout_device(device_id: int):
"""Check out a USB device."""
device = Machine.query.filter_by(
machineid=device_id,
machinetypeid=get_usb_machinetype_id()
).first()
device = USBDevice.query.filter_by(usbdeviceid=device_id, isactive=True).first()
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device with ID {device_id} not found',
http_code=404
)
# Check if already checked out
active_checkout = USBCheckout.query.filter_by(
machineid=device_id,
checkin_time=None
).first()
if active_checkout:
if device.ischeckedout:
return error_response(
ErrorCodes.CONFLICT,
f'Device is already checked out by {active_checkout.sso}',
f'Device is already checked out to {device.currentusername or device.currentuserid}',
http_code=409
)
data = request.get_json() or {}
if not data.get('sso'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'sso is required')
# Create checkout record
checkout = USBCheckout(
machineid=device_id,
usbdeviceid=device_id,
machineid=0, # Legacy field, set to 0 for new checkouts
sso=data['sso'],
checkout_name=data.get('name'),
checkout_reason=data.get('reason'),
checkout_time=datetime.utcnow()
checkout_name=data.get('checkout_name'),
checkout_time=datetime.utcnow(),
checkout_reason=data.get('checkout_reason'),
was_wiped=False
)
# Update device status
device.ischeckedout = True
device.currentuserid = data['sso']
device.currentusername = data.get('checkout_name')
device.currentcheckoutdate = datetime.utcnow()
device.modifieddate = datetime.utcnow()
db.session.add(checkout)
db.session.commit()
return success_response(checkout.to_dict(), message='Device checked out', http_code=201)
@@ -186,90 +281,106 @@ def checkout_device(device_id: int):
@jwt_required()
def checkin_device(device_id: int):
"""Check in a USB device."""
device = Machine.query.filter_by(
machineid=device_id,
machinetypeid=get_usb_machinetype_id()
).first()
device = USBDevice.query.filter_by(usbdeviceid=device_id, isactive=True).first()
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device with ID {device_id} not found',
http_code=404
)
# Find active checkout
active_checkout = USBCheckout.query.filter_by(
machineid=device_id,
checkin_time=None
).first()
if not active_checkout:
if not device.ischeckedout:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'Device is not currently checked out',
http_code=400
)
data = request.get_json() or {}
active_checkout.checkin_time = datetime.utcnow()
active_checkout.was_wiped = data.get('was_wiped', False)
active_checkout.checkin_notes = data.get('notes')
db.session.commit()
return success_response(active_checkout.to_dict(), message='Device checked in')
# Find active checkout
active_checkout = USBCheckout.query.filter_by(
usbdeviceid=device_id,
checkin_time=None
).first()
data = request.get_json() or {}
if active_checkout:
active_checkout.checkin_time = datetime.utcnow()
active_checkout.checkin_notes = data.get('checkin_notes', active_checkout.checkin_notes)
active_checkout.was_wiped = data.get('was_wiped', False)
# Update device status
device.ischeckedout = False
device.currentuserid = None
device.currentusername = None
device.currentcheckoutdate = None
device.modifieddate = datetime.utcnow()
db.session.commit()
return success_response(
active_checkout.to_dict() if active_checkout else None,
message='Device checked in'
)
# =============================================================================
# Checkout History
# =============================================================================
@usb_bp.route('/<int:device_id>/history', methods=['GET'])
@jwt_required()
def get_checkout_history(device_id: int):
@jwt_required(optional=True)
def get_device_history(device_id: int):
"""Get checkout history for a USB device."""
page, per_page = get_pagination_params(request)
query = USBCheckout.query.filter_by(
machineid=device_id
usbdeviceid=device_id
).order_by(USBCheckout.checkout_time.desc())
items, total = paginate_query(query, page, per_page)
data = [c.to_dict() for c in items]
return paginated_response(data, page, per_page, total)
@usb_bp.route('/checkouts', methods=['GET'])
@jwt_required()
@jwt_required(optional=True)
def list_all_checkouts():
"""List all checkouts (active and historical)."""
"""
List all checkouts (active and historical).
Query parameters:
- active: Filter to only active (not returned) checkouts
- sso: Filter by user SSO
"""
page, per_page = get_pagination_params(request)
query = db.session.query(USBCheckout).join(
Machine, USBCheckout.machineid == Machine.machineid
)
query = USBCheckout.query
# Filter by active only
if request.args.get('active', '').lower() == 'true':
query = query.filter(USBCheckout.checkin_time == None)
# Filter by user
if sso := request.args.get('sso'):
query = query.filter(USBCheckout.sso == sso)
query = query.order_by(USBCheckout.checkout_time.desc())
items, total = paginate_query(query, page, per_page)
# Include device info
data = []
for checkout in items:
device = Machine.query.get(checkout.machineid)
item = checkout.to_dict()
item['device'] = {
'machineid': device.machineid,
'alias': device.alias,
'serialnumber': device.serialnumber
} if device else None
data.append(item)
data = [c.to_dict() for c in items]
return paginated_response(data, page, per_page, total)
@usb_bp.route('/checkouts/active', methods=['GET'])
@jwt_required(optional=True)
def list_active_checkouts():
"""List all currently active checkouts."""
checkouts = USBCheckout.query.filter(
USBCheckout.checkin_time == None
).order_by(USBCheckout.checkout_time.desc()).all()
return success_response([c.to_dict() for c in checkouts])

4
plugins/usb/models/__init__.py
View File

@@ -1,5 +1,5 @@
"""USB plugin models."""
from .usb_checkout import USBCheckout
from .usb_device import USBDevice, USBDeviceType, USBCheckout
__all__ = ['USBCheckout']
__all__ = ['USBDevice', 'USBDeviceType', 'USBCheckout']

38
plugins/usb/models/usb_checkout.py
View File

@@ -1,38 +0,0 @@
"""USB Checkout model."""
from shopdb.extensions import db
from datetime import datetime
class USBCheckout(db.Model):
"""
USB device checkout tracking.
References machines table (USB devices have machinetypeid=44).
"""
__tablename__ = 'usbcheckouts'
checkoutid = db.Column(db.Integer, primary_key=True)
machineid = db.Column(db.Integer, db.ForeignKey('machines.machineid'), nullable=False, index=True)
sso = db.Column(db.String(20), nullable=False, index=True)
checkout_name = db.Column(db.String(100))
checkout_reason = db.Column(db.Text)
checkout_time = db.Column(db.DateTime, default=datetime.utcnow, index=True)
checkin_time = db.Column(db.DateTime, index=True)
was_wiped = db.Column(db.Boolean, default=False)
checkin_notes = db.Column(db.Text)
def to_dict(self):
"""Convert to dictionary."""
return {
'checkoutid': self.checkoutid,
'machineid': self.machineid,
'sso': self.sso,
'checkout_name': self.checkout_name,
'checkout_reason': self.checkout_reason,
'checkout_time': self.checkout_time.isoformat() if self.checkout_time else None,
'checkin_time': self.checkin_time.isoformat() if self.checkin_time else None,
'was_wiped': self.was_wiped,
'checkin_notes': self.checkin_notes,
'is_checked_out': self.checkin_time is None
}

44
plugins/usb/models/usb_device.py
View File

@@ -60,6 +60,9 @@ class USBDevice(BaseModel, AuditMixin):
# 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)
@@ -102,56 +105,51 @@ 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'
usbcheckoutid = db.Column(db.Integer, primary_key=True)
checkoutid = db.Column(db.Integer, primary_key=True)
# Device reference
# Device reference (new column linking to usbdevices table)
usbdeviceid = db.Column(
db.Integer,
db.ForeignKey('usbdevices.usbdeviceid', ondelete='CASCADE'),
nullable=False
nullable=True
)
# Legacy reference to machines table (kept for backward compatibility)
machineid = db.Column(db.Integer, nullable=False)
# User info
userid = db.Column(db.String(50), nullable=False, comment='SSO of user')
username = db.Column(db.String(100), nullable=True, comment='Name of user')
sso = db.Column(db.String(20), nullable=False, comment='SSO of user')
checkout_name = db.Column(db.String(100), nullable=True, comment='Name of user')
# Checkout details
checkoutdate = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
checkindate = db.Column(db.DateTime, nullable=True)
expectedreturndate = db.Column(db.DateTime, nullable=True)
checkout_time = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
checkin_time = db.Column(db.DateTime, nullable=True)
# Metadata
purpose = db.Column(db.String(500), nullable=True, comment='Reason for checkout')
notes = db.Column(db.Text, nullable=True)
checkedoutby = db.Column(db.String(50), nullable=True, comment='Admin who processed checkout')
checkedinby = db.Column(db.String(50), nullable=True, comment='Admin who processed checkin')
checkout_reason = db.Column(db.Text, nullable=True, comment='Reason for checkout')
checkin_notes = db.Column(db.Text, nullable=True)
was_wiped = db.Column(db.Boolean, nullable=True, comment='Was device wiped after return')
# Relationships
device = db.relationship('USBDevice', backref=db.backref('checkouts', lazy='dynamic'))
# Indexes
__table_args__ = (
db.Index('idx_usbcheckout_device', 'usbdeviceid'),
db.Index('idx_usbcheckout_user', 'userid'),
db.Index('idx_usbcheckout_dates', 'checkoutdate', 'checkindate'),
)
def __repr__(self):
return f"<USBCheckout device={self.usbdeviceid} user={self.userid}>"
return f"<USBCheckout device={self.usbdeviceid} user={self.sso}>"
@property
def is_active(self):
"""Check if this checkout is currently active (not returned)."""
return self.checkindate is None
return self.checkin_time is None
@property
def duration_days(self):
"""Get duration of checkout in days."""
end = self.checkindate or datetime.utcnow()
delta = end - self.checkoutdate
end = self.checkin_time or datetime.utcnow()
delta = end - self.checkout_time
return delta.days
def to_dict(self):

12
plugins/usb/plugin.py
View File

@@ -10,7 +10,7 @@ from flask import Flask, Blueprint
from shopdb.plugins.base import BasePlugin, PluginMeta
from shopdb.extensions import db
from .models import USBCheckout
from .models import USBDevice, USBDeviceType, USBCheckout
from .api import usb_bp
logger = logging.getLogger(__name__)
@@ -18,10 +18,10 @@ logger = logging.getLogger(__name__)
class USBPlugin(BasePlugin):
"""
USB plugin - manages USB device checkouts.
USB devices are stored in the machines table (machinetypeid=44).
This plugin provides checkout/checkin tracking.
USB plugin - manages USB device tracking and checkouts.
Standalone plugin for tracking USB flash drives, external drives,
and other portable storage devices with checkout/checkin functionality.
"""
def __init__(self):
@@ -54,7 +54,7 @@ class USBPlugin(BasePlugin):
def get_models(self) -> List[Type]:
"""Return list of SQLAlchemy model classes."""
return [USBCheckout]
return [USBDeviceType, USBDevice, USBCheckout]
def init_app(self, app: Flask, db_instance) -> None:
"""Initialize plugin with Flask app."""