Complete USB self-hosted mode (app tables, no external cmmc_usb)
"Create the tables here" for USB now actually works. Add selfhosted.py backing every USB endpoint with the app-owned usbdevices/usbcheckouts tables and returning the same response shape as the external cmmc_usb path. Each route in routes.py delegates to it when usb_directory_mode=selfhosted (the default). - Device status derives from ischeckedout + isactive (retired); check-in/out history is synthesized from usbcheckouts rows (check-out + check-in events); holder names resolve from the self-hosted employee directory. - External cmmc_usb mode unchanged, for sites already running that solution. Verified E2E: create -> list -> checkout -> checkin -> history all correct against the app tables. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
cproudlock
260
plugins/usb/api/selfhosted.py
Normal file
260
plugins/usb/api/selfhosted.py
Normal file
@@ -0,0 +1,260 @@
|
||||
"""Self-hosted USB check-in/out, backed by the app-owned tables.
|
||||
|
||||
Used when usb_directory_mode=selfhosted: the USB routes call these functions
|
||||
instead of the external cmmc_usb database. Data lives in the app tables
|
||||
(usbdevices, usbcheckouts, usbdevicetypes); responses match the same shape the
|
||||
external path returns so the frontend is identical in both modes.
|
||||
|
||||
Field mapping (response contract <- app model):
|
||||
device_id <- USBDevice.serialnumber
|
||||
device_desc <- USBDevice.label
|
||||
status <- derived: retired (not isactive) / checked-out / checked-in
|
||||
locker_location <- USBDevice.storagelocation
|
||||
current_holder <- USBDevice.currentuserid (SSO)
|
||||
checkout_time <- USBDevice.currentcheckoutdate
|
||||
History rows are synthesized from usbcheckouts (each row = a check-out event and,
|
||||
if returned, a check-in event).
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from shopdb.api import (
|
||||
db, success_response, error_response, ErrorCodes,
|
||||
get_pagination_params, paginated_response,
|
||||
)
|
||||
|
||||
from ..models import USBDevice, USBCheckout
|
||||
|
||||
STATUS_CHECKED_IN = 'checked-in'
|
||||
STATUS_CHECKED_OUT = 'checked-out'
|
||||
STATUS_RETIRED = 'retired'
|
||||
_FILTER = {'available': STATUS_CHECKED_IN, 'checkedout': STATUS_CHECKED_OUT, 'retired': STATUS_RETIRED}
|
||||
|
||||
|
||||
def _resolve_name(sso):
|
||||
"""Best-effort name from the self-hosted employee directory."""
|
||||
try:
|
||||
from plugins.employees.models import DirectoryEmployee
|
||||
if sso and str(sso).isdigit():
|
||||
emp = DirectoryEmployee.query.get(int(sso))
|
||||
if emp:
|
||||
return f'{emp.firstname} {emp.lastname}'.strip()
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def _status(device):
|
||||
if not device.isactive:
|
||||
return STATUS_RETIRED
|
||||
return STATUS_CHECKED_OUT if device.ischeckedout else STATUS_CHECKED_IN
|
||||
|
||||
|
||||
def _device_dict(device):
|
||||
return {
|
||||
'device_id': device.serialnumber,
|
||||
'device_desc': device.label,
|
||||
'device_owner': None,
|
||||
'owner_name': None,
|
||||
'status': _status(device),
|
||||
'locker_location': device.storagelocation,
|
||||
'current_holder': device.currentuserid,
|
||||
'current_holder_name': device.currentusername,
|
||||
'checkout_time': device.currentcheckoutdate.isoformat() + 'Z' if device.currentcheckoutdate else None,
|
||||
}
|
||||
|
||||
|
||||
def _checkout_events(checkout):
|
||||
"""One usbcheckouts row -> its check-out event (+ check-in event if returned)."""
|
||||
name = checkout.checkoutname or _resolve_name(checkout.sso)
|
||||
events = [{
|
||||
'log_id': checkout.checkoutid * 2,
|
||||
'badge_number': checkout.sso,
|
||||
'badge_name': name,
|
||||
'device_id': checkout.device.serialnumber if checkout.device else None,
|
||||
'action': 'check-out',
|
||||
'timestamp': checkout.checkouttime.isoformat() + 'Z' if checkout.checkouttime else None,
|
||||
'scanned_viruses': None,
|
||||
'locker_location': None,
|
||||
'sanitized': None,
|
||||
}]
|
||||
if checkout.checkintime:
|
||||
events.append({
|
||||
'log_id': checkout.checkoutid * 2 + 1,
|
||||
'badge_number': checkout.sso,
|
||||
'badge_name': name,
|
||||
'device_id': checkout.device.serialnumber if checkout.device else None,
|
||||
'action': 'check-in',
|
||||
'timestamp': checkout.checkintime.isoformat() + 'Z',
|
||||
'scanned_viruses': None,
|
||||
'locker_location': None,
|
||||
'sanitized': checkout.waswiped,
|
||||
})
|
||||
return events
|
||||
|
||||
|
||||
def _find(device_id):
|
||||
return USBDevice.query.filter_by(serialnumber=device_id).first()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def list_devices(request):
|
||||
page, per_page = get_pagination_params(request)
|
||||
query = USBDevice.query
|
||||
status_arg = request.args.get('status', '').strip().lower()
|
||||
if status_arg in _FILTER:
|
||||
want = _FILTER[status_arg]
|
||||
if want == STATUS_RETIRED:
|
||||
query = query.filter_by(isactive=False)
|
||||
elif want == STATUS_CHECKED_OUT:
|
||||
query = query.filter_by(isactive=True, ischeckedout=True)
|
||||
else:
|
||||
query = query.filter_by(isactive=True, ischeckedout=False)
|
||||
search = request.args.get('search', '').strip()
|
||||
if search:
|
||||
like = f'%{search}%'
|
||||
query = query.filter(db.or_(USBDevice.serialnumber.ilike(like),
|
||||
USBDevice.label.ilike(like)))
|
||||
total = query.count()
|
||||
rows = (query.order_by(USBDevice.serialnumber)
|
||||
.limit(per_page).offset((page - 1) * per_page).all())
|
||||
return paginated_response([_device_dict(d) for d in rows], page, per_page, total)
|
||||
|
||||
|
||||
def get_device(device_id):
|
||||
device = _find(device_id)
|
||||
if not device:
|
||||
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
||||
result = _device_dict(device)
|
||||
checkouts = (USBCheckout.query.filter_by(usbdeviceid=device.usbdeviceid)
|
||||
.order_by(USBCheckout.checkouttime.desc()).limit(20).all())
|
||||
logs = []
|
||||
for checkout in checkouts:
|
||||
logs.extend(_checkout_events(checkout))
|
||||
logs.sort(key=lambda item: item['timestamp'] or '', reverse=True)
|
||||
result['checkinoutlog'] = logs[:20]
|
||||
return success_response(result)
|
||||
|
||||
|
||||
def create_device(data):
|
||||
device_id = (data.get('device_id') or '').strip()
|
||||
if not device_id:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'device_id is required')
|
||||
if _find(device_id):
|
||||
return error_response(ErrorCodes.CONFLICT, 'device_id already exists', http_code=409)
|
||||
device = USBDevice(serialnumber=device_id, label=data.get('device_desc'),
|
||||
storagelocation=data.get('locker_location'),
|
||||
ischeckedout=False, isactive=True)
|
||||
db.session.add(device)
|
||||
db.session.commit()
|
||||
return success_response(_device_dict(device), message='Device created', http_code=201)
|
||||
|
||||
|
||||
def update_device(device_id, data):
|
||||
device = _find(device_id)
|
||||
if not device:
|
||||
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
||||
if 'device_desc' in data:
|
||||
device.label = data['device_desc']
|
||||
if 'locker_location' in data:
|
||||
device.storagelocation = data['locker_location']
|
||||
if 'status' in data:
|
||||
if data['status'] == STATUS_RETIRED:
|
||||
device.isactive = False
|
||||
elif data['status'] in (STATUS_CHECKED_IN, STATUS_CHECKED_OUT):
|
||||
device.isactive = True
|
||||
device.ischeckedout = data['status'] == STATUS_CHECKED_OUT
|
||||
db.session.commit()
|
||||
return success_response(_device_dict(device), message='Device updated')
|
||||
|
||||
|
||||
def retire_device(device_id):
|
||||
device = _find(device_id)
|
||||
if not device:
|
||||
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
||||
device.isactive = False
|
||||
db.session.commit()
|
||||
return success_response(_device_dict(device), message='Device retired')
|
||||
|
||||
|
||||
def checkout_device(device_id, data):
|
||||
badge = (data.get('badge') or '').strip()
|
||||
if not badge:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'badge is required')
|
||||
device = _find(device_id)
|
||||
if not device:
|
||||
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
||||
if device.ischeckedout:
|
||||
return error_response(ErrorCodes.CONFLICT, 'Device is already checked out', http_code=409)
|
||||
name = _resolve_name(badge)
|
||||
now = datetime.utcnow()
|
||||
db.session.add(USBCheckout(usbdeviceid=device.usbdeviceid, machineid=0, sso=badge,
|
||||
checkoutname=name, checkouttime=now,
|
||||
checkoutreason=data.get('reason')))
|
||||
device.ischeckedout = True
|
||||
device.currentuserid = badge
|
||||
device.currentusername = name
|
||||
device.currentcheckoutdate = now
|
||||
if data.get('locker_location'):
|
||||
device.storagelocation = data['locker_location']
|
||||
db.session.commit()
|
||||
return success_response(_device_dict(device), message='Device checked out', http_code=201)
|
||||
|
||||
|
||||
def checkin_device(device_id, data):
|
||||
badge = (data.get('badge') or '').strip()
|
||||
if not badge:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'badge is required')
|
||||
device = _find(device_id)
|
||||
if not device:
|
||||
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
||||
if not device.ischeckedout:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Device is not checked out')
|
||||
open_checkout = (USBCheckout.query
|
||||
.filter_by(usbdeviceid=device.usbdeviceid, checkintime=None)
|
||||
.order_by(USBCheckout.checkouttime.desc()).first())
|
||||
if open_checkout:
|
||||
open_checkout.checkintime = datetime.utcnow()
|
||||
open_checkout.waswiped = bool(data.get('sanitized'))
|
||||
open_checkout.checkinnotes = data.get('notes')
|
||||
device.ischeckedout = False
|
||||
device.currentuserid = None
|
||||
device.currentusername = None
|
||||
device.currentcheckoutdate = None
|
||||
if data.get('locker_location'):
|
||||
device.storagelocation = data['locker_location']
|
||||
db.session.commit()
|
||||
return success_response(_device_dict(device), message='Device checked in')
|
||||
|
||||
|
||||
def device_history(device_id, request):
|
||||
device = _find(device_id)
|
||||
if not device:
|
||||
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
||||
page, per_page = get_pagination_params(request)
|
||||
checkouts = (USBCheckout.query.filter_by(usbdeviceid=device.usbdeviceid)
|
||||
.order_by(USBCheckout.checkouttime.desc()).all())
|
||||
logs = []
|
||||
for checkout in checkouts:
|
||||
logs.extend(_checkout_events(checkout))
|
||||
logs.sort(key=lambda item: item['timestamp'] or '', reverse=True)
|
||||
total = len(logs)
|
||||
start = (page - 1) * per_page
|
||||
return paginated_response(logs[start:start + per_page], page, per_page, total)
|
||||
|
||||
|
||||
def list_checkouts(request, active_only):
|
||||
query = USBCheckout.query
|
||||
if active_only:
|
||||
query = query.filter_by(checkintime=None)
|
||||
badge = request.args.get('badge', '').strip()
|
||||
if badge:
|
||||
query = query.filter_by(sso=badge)
|
||||
rows = query.order_by(USBCheckout.checkouttime.desc()).all()
|
||||
data = []
|
||||
for checkout in rows:
|
||||
event = _checkout_events(checkout)[0]
|
||||
event['device_status'] = _status(checkout.device) if checkout.device else None
|
||||
data.append(event)
|
||||
return success_response(data)
|
||||
Reference in New Issue
Block a user