64a5abdb08
The measuringtools plugin was missing from two cross-cutting surfaces: the asset-identifier matrix (no measuring_tool column or per-type keys - gauge lab reference is their primary identifier) and global search (results fell to a generic URL and gaugelabreference was never searched). Measuring tools now have identifier toggles, gated gauge-lab and maintenance-reference fields on their form and detail, a search domain toggle, gage-tag search, and proper labels, routes, and filter chips in search results. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
370 lines
15 KiB
Python
370 lines
15 KiB
Python
"""Measuring-tools plugin API.
|
|
|
|
Two resources: the measuringtooltypes lookup (CRUD with an in-use delete guard)
|
|
and the measuring tools themselves (an Asset core row plus a measuringtools
|
|
extension row, written in one payload). Reads are jwt-optional per the app
|
|
convention; writes require the measuringtools.* permissions.
|
|
|
|
Calibration status is derived at read time in the model (see
|
|
models.derive_status), never stored, so the report and list badges are always
|
|
current.
|
|
"""
|
|
|
|
from datetime import date, datetime
|
|
|
|
from flask import Blueprint, request
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
from shopdb.api import (
|
|
db, Asset, AssetType, AuditLog,
|
|
success_response, error_response, paginated_response, ErrorCodes,
|
|
get_pagination_params, paginate_query,
|
|
require_permission, apply_import_timestamps,
|
|
)
|
|
|
|
from ..models import MeasuringTool, MeasuringToolType, derive_status, STATUS_COLORS
|
|
|
|
measuringtools_bp = Blueprint('measuringtools', __name__)
|
|
|
|
|
|
def _parse_date(value):
|
|
"""Accept 'YYYY-MM-DD' (or None/empty) -> date or None."""
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.strptime(str(value)[:10], '%Y-%m-%d').date()
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _merged(tool, today=None):
|
|
"""Asset core dict with the measuringtools extension nested under 'measuringtool'."""
|
|
result = tool.asset.to_dict() if tool.asset else {}
|
|
result['measuringtool'] = tool.to_dict(today)
|
|
return result
|
|
|
|
|
|
# =============================================================================
|
|
# Measuring-tool types
|
|
# =============================================================================
|
|
|
|
@measuringtools_bp.route('/types', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_types():
|
|
"""List measuring-tool types."""
|
|
page, per_page = get_pagination_params(request)
|
|
query = MeasuringToolType.query
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(MeasuringToolType.isactive == True)
|
|
if search := request.args.get('search'):
|
|
query = query.filter(MeasuringToolType.name.ilike(f'%{search}%'))
|
|
query = query.order_by(MeasuringToolType.name)
|
|
items, total = paginate_query(query, page, per_page)
|
|
return paginated_response([t.to_dict() for t in items], page, per_page, total)
|
|
|
|
|
|
@measuringtools_bp.route('/types/<int:type_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_type(type_id: int):
|
|
"""Get one measuring-tool type."""
|
|
tool_type = db.session.get(MeasuringToolType, type_id)
|
|
if not tool_type:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'Measuring-tool type {type_id} not found', http_code=404)
|
|
return success_response(tool_type.to_dict())
|
|
|
|
|
|
@measuringtools_bp.route('/types', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('measuringtools.create')
|
|
def create_type():
|
|
"""Create a measuring-tool type (reactivates a soft-deleted same-named one)."""
|
|
data = request.get_json() or {}
|
|
name = (data.get('name') or '').strip()
|
|
if not name:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'name is required')
|
|
|
|
existing = MeasuringToolType.query.filter_by(name=name).first()
|
|
if existing:
|
|
if not existing.isactive:
|
|
existing.isactive = True
|
|
for key in ('description', 'color'):
|
|
if data.get(key) is not None:
|
|
setattr(existing, key, data[key])
|
|
db.session.commit()
|
|
return success_response(existing.to_dict(), message='Reactivated existing type')
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Measuring-tool type '{name}' already exists", http_code=409)
|
|
|
|
tool_type = MeasuringToolType(
|
|
name=name,
|
|
description=data.get('description'),
|
|
color=data.get('color'),
|
|
)
|
|
db.session.add(tool_type)
|
|
db.session.commit()
|
|
return success_response(tool_type.to_dict(), message='Measuring-tool type created',
|
|
http_code=201)
|
|
|
|
|
|
@measuringtools_bp.route('/types/<int:type_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('measuringtools.edit')
|
|
def update_type(type_id: int):
|
|
"""Update a measuring-tool type."""
|
|
tool_type = db.session.get(MeasuringToolType, type_id)
|
|
if not tool_type:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'Measuring-tool type {type_id} not found', http_code=404)
|
|
data = request.get_json() or {}
|
|
if 'name' in data and data['name'] != tool_type.name:
|
|
if MeasuringToolType.query.filter_by(name=data['name']).first():
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Measuring-tool type '{data['name']}' already exists",
|
|
http_code=409)
|
|
for key in ('name', 'description', 'color', 'isactive'):
|
|
if key in data:
|
|
setattr(tool_type, key, data[key])
|
|
db.session.commit()
|
|
return success_response(tool_type.to_dict(), message='Measuring-tool type updated')
|
|
|
|
|
|
@measuringtools_bp.route('/types/<int:type_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('measuringtools.delete')
|
|
def delete_type(type_id: int):
|
|
"""Delete a measuring-tool type. Refused if any tool still uses it."""
|
|
tool_type = db.session.get(MeasuringToolType, type_id)
|
|
if not tool_type:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'Measuring-tool type {type_id} not found', http_code=404)
|
|
inuse = MeasuringTool.query.filter_by(measuringtooltypeid=type_id).count()
|
|
if inuse:
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f'Cannot delete: {inuse} tool(s) still use this type',
|
|
http_code=409)
|
|
db.session.delete(tool_type)
|
|
db.session.commit()
|
|
return success_response(message='Measuring-tool type deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Measuring tools
|
|
# =============================================================================
|
|
|
|
@measuringtools_bp.route('', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_tools():
|
|
"""List measuring tools with filters + pagination.
|
|
|
|
Query params: typeid, locationid, statusid, calibrationstatus, search,
|
|
active, page, perpage.
|
|
"""
|
|
page, per_page = get_pagination_params(request)
|
|
query = db.session.query(MeasuringTool).join(Asset)
|
|
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(Asset.isactive == True)
|
|
# Exact-match natural-key lookup for idempotent import (asset number).
|
|
if exactassetnumber := request.args.get('assetnumber'):
|
|
query = query.filter(Asset.assetnumber == exactassetnumber)
|
|
if search := request.args.get('search'):
|
|
query = query.filter(db.or_(
|
|
Asset.assetnumber.ilike(f'%{search}%'),
|
|
Asset.name.ilike(f'%{search}%'),
|
|
Asset.serialnumber.ilike(f'%{search}%'),
|
|
))
|
|
if type_id := request.args.get('typeid', type=int):
|
|
query = query.filter(MeasuringTool.measuringtooltypeid == type_id)
|
|
if location_id := request.args.get('locationid', type=int):
|
|
query = query.filter(Asset.locationid == location_id)
|
|
if status_id := request.args.get('statusid', type=int):
|
|
query = query.filter(Asset.statusid == status_id)
|
|
|
|
query = query.order_by(Asset.assetnumber)
|
|
items, total = paginate_query(query, page, per_page)
|
|
|
|
today = date.today()
|
|
data = [_merged(tool, today) for tool in items]
|
|
|
|
# Calibration status is derived, so it cannot be a SQL filter - apply it
|
|
# to the built rows. Filtering post-pagination is acceptable here because
|
|
# the dataset is a gage lab's worth of tools, not a fleet.
|
|
calibrationstatus = request.args.get('calibrationstatus')
|
|
if calibrationstatus:
|
|
data = [d for d in data
|
|
if d['measuringtool']['calibrationstatus'] == calibrationstatus]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@measuringtools_bp.route('/<int:tool_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_tool(tool_id: int):
|
|
"""Get one measuring tool (asset core merged with the extension)."""
|
|
tool = db.session.get(MeasuringTool, tool_id)
|
|
if not tool:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'Measuring tool {tool_id} not found', http_code=404)
|
|
return success_response(_merged(tool))
|
|
|
|
|
|
@measuringtools_bp.route('/by-asset/<int:asset_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_tool_by_asset(asset_id: int):
|
|
"""Get a measuring tool by its core asset id."""
|
|
tool = MeasuringTool.query.filter_by(assetid=asset_id).first()
|
|
if not tool:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'Measuring tool for asset {asset_id} not found', http_code=404)
|
|
return success_response(_merged(tool))
|
|
|
|
|
|
def _asset_type_id():
|
|
"""Resolve the seeded 'measuring_tool' asset-type id, or None."""
|
|
asset_type = AssetType.query.filter_by(assettype='measuring_tool').first()
|
|
return asset_type.assettypeid if asset_type else None
|
|
|
|
|
|
@measuringtools_bp.route('', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('measuringtools.create')
|
|
def create_tool():
|
|
"""Create a measuring tool: one Asset core row + one extension row."""
|
|
data = request.get_json() or {}
|
|
if not data.get('assetnumber'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'assetnumber is required')
|
|
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Asset with number '{data['assetnumber']}' already exists",
|
|
http_code=409)
|
|
|
|
assettypeid = _asset_type_id()
|
|
if not assettypeid:
|
|
return error_response(ErrorCodes.INTERNAL_ERROR,
|
|
'measuring_tool asset type not found. Plugin may not be '
|
|
'properly installed.', http_code=500)
|
|
|
|
asset = Asset(
|
|
assetnumber=data['assetnumber'],
|
|
name=data.get('name'),
|
|
gaugelabreference=data.get('gaugelabreference'),
|
|
maintenancereference=data.get('maintenancereference'),
|
|
serialnumber=data.get('serialnumber'),
|
|
assettypeid=assettypeid,
|
|
statusid=data.get('statusid', 1),
|
|
locationid=data.get('locationid'),
|
|
businessunitid=data.get('businessunitid'),
|
|
mapx=data.get('mapx'),
|
|
mapy=data.get('mapy'),
|
|
notes=data.get('notes'),
|
|
)
|
|
db.session.add(asset)
|
|
db.session.flush() # assign assetid
|
|
|
|
tool = MeasuringTool(
|
|
assetid=asset.assetid,
|
|
measuringtooltypeid=data.get('measuringtooltypeid'),
|
|
calibrationintervaldays=data.get('calibrationintervaldays'),
|
|
lastcalibrationdate=_parse_date(data.get('lastcalibrationdate')),
|
|
nextcalibrationdate=_parse_date(data.get('nextcalibrationdate')),
|
|
calibrationprovider=(data.get('calibrationprovider') or '').strip() or None,
|
|
notes=(data.get('notes') or '').strip() or None,
|
|
)
|
|
db.session.add(tool)
|
|
db.session.flush()
|
|
|
|
apply_import_timestamps(asset, data)
|
|
AuditLog.log('created', 'MeasuringTool', entityid=tool.measuringtoolid,
|
|
entityname=asset.assetnumber)
|
|
db.session.commit()
|
|
return success_response(_merged(tool), message='Measuring tool created', http_code=201)
|
|
|
|
|
|
# Asset core fields writable through this plugin's write path.
|
|
_ASSET_FIELDS = ('assetnumber', 'name', 'gaugelabreference',
|
|
'maintenancereference', 'serialnumber',
|
|
'statusid', 'locationid', 'businessunitid', 'mapx', 'mapy',
|
|
'notes', 'isactive')
|
|
# Extension fields with plain assignment (dates handled separately).
|
|
_TOOL_FIELDS = ('measuringtooltypeid', 'calibrationintervaldays',
|
|
'calibrationprovider', 'notes')
|
|
|
|
|
|
@measuringtools_bp.route('/<int:tool_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('measuringtools.edit')
|
|
def update_tool(tool_id: int):
|
|
"""Update a measuring tool (asset core fields + extension in one payload)."""
|
|
tool = db.session.get(MeasuringTool, tool_id)
|
|
if not tool:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'Measuring tool {tool_id} not found', http_code=404)
|
|
data = request.get_json() or {}
|
|
asset = tool.asset
|
|
|
|
if 'assetnumber' in data and data['assetnumber'] != asset.assetnumber:
|
|
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Asset with number '{data['assetnumber']}' already exists",
|
|
http_code=409)
|
|
|
|
changes = {}
|
|
for key in _ASSET_FIELDS:
|
|
if key in data:
|
|
if getattr(asset, key) != data[key]:
|
|
changes[key] = {'old': getattr(asset, key), 'new': data[key]}
|
|
setattr(asset, key, data[key])
|
|
for key in _TOOL_FIELDS:
|
|
if key in data:
|
|
if getattr(tool, key) != data[key]:
|
|
changes[key] = {'old': getattr(tool, key), 'new': data[key]}
|
|
setattr(tool, key, data[key])
|
|
for key in ('lastcalibrationdate', 'nextcalibrationdate'):
|
|
if key in data:
|
|
setattr(tool, key, _parse_date(data[key]))
|
|
|
|
if changes:
|
|
AuditLog.log('updated', 'MeasuringTool', entityid=tool.measuringtoolid,
|
|
entityname=asset.assetnumber, changes=changes)
|
|
apply_import_timestamps(asset, data)
|
|
db.session.commit()
|
|
return success_response(_merged(tool), message='Measuring tool updated')
|
|
|
|
|
|
@measuringtools_bp.route('/<int:tool_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('measuringtools.delete')
|
|
def delete_tool(tool_id: int):
|
|
"""Soft delete a measuring tool (deactivates its asset)."""
|
|
tool = db.session.get(MeasuringTool, tool_id)
|
|
if not tool:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'Measuring tool {tool_id} not found', http_code=404)
|
|
tool.asset.isactive = False
|
|
AuditLog.log('deleted', 'MeasuringTool', entityid=tool.measuringtoolid,
|
|
entityname=tool.asset.assetnumber)
|
|
db.session.commit()
|
|
return success_response(message='Measuring tool deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Calibration report (for the Reports hub)
|
|
# =============================================================================
|
|
|
|
@measuringtools_bp.route('/report/calibration', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def calibration_report():
|
|
"""Counts + lists bucketed by derived calibration status."""
|
|
today = date.today()
|
|
buckets = {'overdue': [], 'duesoon': [], 'current': [], 'unknown': []}
|
|
query = db.session.query(MeasuringTool).join(Asset).filter(Asset.isactive == True)
|
|
for tool in query.all():
|
|
buckets.setdefault(derive_status(tool.nextcalibrationdate, today), []).append(
|
|
_merged(tool, today))
|
|
return success_response({
|
|
'counts': {key: len(value) for key, value in buckets.items()},
|
|
'buckets': buckets,
|
|
'statuscolors': STATUS_COLORS,
|
|
})
|