Asset-side printer save, status CRUD, real printer types, network fix

- Printers save through the asset blueprint (PUT /printers) instead of the
  legacy machines API; restrict supply-model picker to printer models.
- Asset statuses get full CRUD (PUT/DELETE with in-use guard); canonical set.
- Printer types reseeded to a real classification set + list filter.
- Equipment accepts gauge/maintenance references.
- Fix network list emitting network_device instead of networkdevice (View 404).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-06-26 08:35:32 -04:00
parent 4626280dc4
commit 0436e8b0af
8 changed files with 451 additions and 39 deletions

View File

@@ -282,6 +282,8 @@ def create_equipment():
asset = Asset( asset = Asset(
assetnumber=data['assetnumber'], assetnumber=data['assetnumber'],
name=data.get('name'), name=data.get('name'),
gaugelabreference=data.get('gaugelabreference'),
maintenancereference=data.get('maintenancereference'),
serialnumber=data.get('serialnumber'), serialnumber=data.get('serialnumber'),
assettypeid=equipment_type.assettypeid, assettypeid=equipment_type.assettypeid,
statusid=data.get('statusid', 1), statusid=data.get('statusid', 1),
@@ -357,8 +359,10 @@ def update_equipment(equipment_id: int):
changes = {} changes = {}
# Update asset fields # Update asset fields
asset_fields = ['assetnumber', 'name', 'serialnumber', 'statusid', asset_fields = ['assetnumber', 'name', 'gaugelabreference',
'locationid', 'businessunitid', 'mapx', 'mapy', 'notes', 'isactive'] 'maintenancereference', 'serialnumber', 'statusid',
'locationid', 'businessunitid', 'mapx', 'mapy',
'notes', 'isactive']
for key in asset_fields: for key in asset_fields:
if key in data: if key in data:
old_val = getattr(asset, key) old_val = getattr(asset, key)

View File

@@ -207,7 +207,7 @@ def list_network_devices():
data = [] data = []
for netdev in items: for netdev in items:
item = netdev.asset.to_dict() if netdev.asset else {} item = netdev.asset.to_dict() if netdev.asset else {}
item['network_device'] = netdev.to_dict() item['networkdevice'] = netdev.to_dict()
data.append(item) data.append(item)
return paginated_response(data, page, per_page, total) return paginated_response(data, page, per_page, total)

View File

@@ -15,8 +15,15 @@ from shopdb.utils.responses import (
) )
from shopdb.utils.pagination import get_pagination_params, paginate_query from shopdb.utils.pagination import get_pagination_params, paginate_query
from ..models import Printer, PrinterType from ..models import Printer, PrinterType, ModelSupply
from ..services import ZabbixService from ..models.model_supply import SUPPLY_TYPES, SUPPLY_COLORS, CAPACITY_TIERS
from ..services import (
ZabbixService,
classifysupply,
derivesupplytype,
derivecolor,
lookupsupplies,
)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -134,8 +141,8 @@ def list_printers():
) )
# Type filter # Type filter
if type_id := request.args.get('type_id'): if typeid := request.args.get('typeid', request.args.get('type_id')):
query = query.filter(Printer.printertypeid == int(type_id)) query = query.filter(Printer.printertypeid == int(typeid))
# Vendor filter # Vendor filter
if vendor_id := request.args.get('vendor_id'): if vendor_id := request.args.get('vendor_id'):
@@ -357,9 +364,10 @@ def update_printer(printer_id: int):
http_code=409 http_code=409
) )
# Update asset fields # Update asset fields (gauge lab / maintenance refs are equipment-only)
asset_fields = ['assetnumber', 'name', 'serialnumber', 'statusid', asset_fields = ['assetnumber', 'name', 'serialnumber', 'statusid',
'locationid', 'businessunitid', 'mapx', 'mapy', 'notes', 'isactive'] 'locationid', 'businessunitid', 'mapx', 'mapy',
'notes', 'isactive']
for key in asset_fields: for key in asset_fields:
if key in data: if key in data:
setattr(asset, key, data[key]) setattr(asset, key, data[key])
@@ -372,6 +380,27 @@ def update_printer(printer_id: int):
if key in data: if key in data:
setattr(printer, key, data[key]) setattr(printer, key, data[key])
# Upsert the primary IP communication when an ipaddress is supplied, so a
# single PUT updates core, extension, and network in one call.
if 'ipaddress' in data:
ip = (data.get('ipaddress') or '').strip()
comm = Communication.query.filter_by(
assetid=asset.assetid, isprimary=True).first()
if ip:
if comm:
comm.ipaddress = ip
else:
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
if ip_comtype:
db.session.add(Communication(
assetid=asset.assetid,
comtypeid=ip_comtype.comtypeid,
ipaddress=ip,
isprimary=True,
))
elif comm:
comm.ipaddress = None
db.session.commit() db.session.commit()
result = asset.to_dict() result = asset.to_dict()
@@ -426,17 +455,25 @@ def get_printer_supplies(printer_id: int):
service = ZabbixService() service = ZabbixService()
if not service.isconfigured or not service.isreachable: if not service.isconfigured or not service.isreachable:
# Return empty supplies if Zabbix not available (fail gracefully) # fail soft when zabbix off or down
return success_response({ return success_response({
'ipaddress': comm.ipaddress, 'ipaddress': comm.ipaddress,
'pingstatus': '-1',
'supplies': [] 'supplies': []
}) })
supplies = service.getsuppliesbyip(comm.ipaddress) # vendor drives waste-cartridge rules; modelnumberid drives part lookup
vendor_name = printer.vendor.vendor if getattr(printer, 'vendor', None) else None
raw_supplies = service.getsuppliesbyip(comm.ipaddress) or []
supplies = [
_annotate_supply(s, vendor_name, printer.modelnumberid) for s in raw_supplies
]
return success_response({ return success_response({
'ipaddress': comm.ipaddress, 'ipaddress': comm.ipaddress,
'supplies': supplies or [] 'pingstatus': service.getpingstatus(comm.ipaddress),
'supplies': supplies
}) })
@@ -444,8 +481,32 @@ def get_printer_supplies(printer_id: int):
# Low Supplies # Low Supplies
# ============================================================================= # =============================================================================
def _annotate_supply(supply, vendor_name, modelnumberid):
"""Add status, remaining percent, and part numbers to a raw supply dict.
Waste cartridge direction depends on vendor, so classification lives in
the supply_parts helper. Part numbers come from the modelsupplies table.
"""
level = supply.get('level', 0)
name = supply.get('name', 'Unknown')
supplytype = derivesupplytype(name)
color = derivecolor(name, supply.get('color'))
cls = classifysupply(level, name, vendor_name)
return {
'name': name,
'level': level,
'color': color,
'supplytype': supplytype,
'status': cls['status'],
'remaining': cls['remaining'],
'iswaste': cls['iswaste'],
'isdrum': cls['isdrum'],
'partnumbers': lookupsupplies(modelnumberid, color, supplytype),
}
def _get_low_supplies_data(): def _get_low_supplies_data():
"""Build low supplies data (cached for 10 minutes).""" """Build low supplies data (cached for 5 minutes)."""
cached = cache.get('printers_low_supplies') cached = cache.get('printers_low_supplies')
if cached is not None: if cached is not None:
return cached return cached
@@ -454,56 +515,51 @@ def _get_low_supplies_data():
if not service.isconfigured or not service.isreachable: if not service.isconfigured or not service.isreachable:
return {'printers': [], 'summary': {'total_checked': 0, 'low': 0, 'critical': 0}} return {'printers': [], 'summary': {'total_checked': 0, 'low': 0, 'critical': 0}}
# All active printers with an IP address # active printers with an IP, with vendor and model for waste/part rules
printers = ( rows = (
db.session.query(Printer, Asset, Communication) db.session.query(Printer, Asset, Communication, Vendor, Model)
.join(Asset, Asset.assetid == Printer.assetid) .join(Asset, Asset.assetid == Printer.assetid)
.join(Communication, Communication.assetid == Asset.assetid) .join(Communication, Communication.assetid == Asset.assetid)
.outerjoin(Vendor, Vendor.vendorid == Printer.vendorid)
.outerjoin(Model, Model.modelnumberid == Printer.modelnumberid)
.filter(Asset.isactive == True) .filter(Asset.isactive == True)
.filter(Communication.ipaddress.isnot(None)) .filter(Communication.ipaddress.isnot(None))
.filter(Communication.ipaddress != '') .filter(Communication.ipaddress != '')
.all() .all()
) )
# Dedupe by printer id (may have multiple comms) # dedupe by printer id (a printer may have several comms)
seen = set() seen = set()
unique_printers = [] unique_printers = []
for printer, asset, comm in printers: for printer, asset, comm, vendor, model in rows:
if printer.printerid not in seen: if printer.printerid not in seen:
seen.add(printer.printerid) seen.add(printer.printerid)
unique_printers.append((printer, asset, comm)) unique_printers.append((printer, asset, comm, vendor, model))
results = [] results = []
total_checked = 0 total_checked = 0
for printer, asset, comm in unique_printers: for printer, asset, comm, vendor, model in unique_printers:
supplies = service.getsuppliesbyip_cached(comm.ipaddress) supplies = service.getsuppliesbyip_cached(comm.ipaddress)
if supplies is None: if supplies is None:
continue continue
total_checked += 1 total_checked += 1
# Annotate each supply with status vendor_name = vendor.vendor if vendor else None
model_number = model.modelnumber if model else None
modelnumberid = model.modelnumberid if model else None
annotated = [] annotated = []
has_low = False has_low = False
for s in supplies: for s in supplies:
level = s.get('level', 0) item = _annotate_supply(s, vendor_name, modelnumberid)
if level <= 5: if item['status'] != 'ok':
status = 'critical'
has_low = True has_low = True
elif level <= 10: annotated.append(item)
status = 'low'
has_low = True
else:
status = 'ok'
annotated.append({
'name': s.get('name', 'Unknown'),
'level': level,
'status': status
})
if has_low: if has_low:
# Get location name # location name for the report row
location_name = None location_name = None
if asset.locationid: if asset.locationid:
from shopdb.core.models import Location from shopdb.core.models import Location
@@ -516,6 +572,8 @@ def _get_low_supplies_data():
'printername': asset.name or printer.hostname or '', 'printername': asset.name or printer.hostname or '',
'assetnumber': asset.assetnumber or '', 'assetnumber': asset.assetnumber or '',
'ipaddress': comm.ipaddress, 'ipaddress': comm.ipaddress,
'vendor': vendor_name,
'model': model_number,
'location': location_name, 'location': location_name,
'supplies': annotated 'supplies': annotated
}) })
@@ -539,7 +597,7 @@ def _get_low_supplies_data():
} }
} }
cache.set('printers_low_supplies', data, timeout=600) cache.set('printers_low_supplies', data, timeout=300)
return data return data
@@ -551,6 +609,62 @@ def low_supplies():
return success_response(data) return success_response(data)
@printers_asset_bp.route('/lookup', methods=['GET'])
@jwt_required(optional=True)
def printer_lookup():
"""Find a printer by IP or FQDN. Parity with the classic printerlookup.asp.
Zabbix uses this to jump straight to a printer record. Query with
?ip=x.x.x.x or ?fqdn=hostname; returns the matching printer id.
"""
ip = (request.args.get('ip') or '').strip()
fqdn = (request.args.get('fqdn') or '').strip()
lookup_value = ip or fqdn
if not lookup_value:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'Provide ip or fqdn'
)
# match the IP against any active printer communication
row = (
db.session.query(Printer, Asset)
.join(Asset, Asset.assetid == Printer.assetid)
.join(Communication, Communication.assetid == Asset.assetid)
.filter(Asset.isactive == True)
.filter(Communication.ipaddress == lookup_value)
.first()
)
if not row:
return error_response(
ErrorCodes.NOT_FOUND,
f'Printer not found: {lookup_value}',
http_code=404
)
printer, asset = row
return success_response({
'printerid': printer.printerid,
'assetid': asset.assetid,
'assetnumber': asset.assetnumber,
'name': asset.name or printer.hostname,
})
@printers_asset_bp.route('/supplies/refresh', methods=['POST'])
@jwt_required()
def refresh_supplies_cache():
"""Clear cached Zabbix supply data so the next read pulls fresh values.
Backs the toner report Refresh button (parity with adminclearcache.asp
type=zabbix).
"""
ZabbixService().clearcache()
return success_response(message='Supply cache cleared')
# ============================================================================= # =============================================================================
# Dashboard # Dashboard
# ============================================================================= # =============================================================================
@@ -605,3 +719,213 @@ def dashboard_summary():
'bytype': [{'type': t, 'count': c} for t, c in by_type], 'bytype': [{'type': t, 'count': c} for t, c in by_type],
'byvendor': [{'vendor': v, 'count': c} for v, c in by_vendor], 'byvendor': [{'vendor': v, 'count': c} for v, c in by_vendor],
}) })
# =============================================================================
# Model Supplies (data-driven toner/drum/waste part numbers)
# =============================================================================
def _validate_supply_payload(data):
"""Return an error message if the supply payload is invalid, else None."""
if not data:
return 'No data provided'
if not data.get('partnumber'):
return 'partnumber is required'
supplytype = data.get('supplytype', 'toner')
if supplytype not in SUPPLY_TYPES:
return f"supplytype must be one of {', '.join(SUPPLY_TYPES)}"
color = data.get('color', 'none')
if color not in SUPPLY_COLORS:
return f"color must be one of {', '.join(SUPPLY_COLORS)}"
capacitytier = data.get('capacitytier', 'standard')
if capacitytier not in CAPACITY_TIERS:
return f"capacitytier must be one of {', '.join(CAPACITY_TIERS)}"
return None
@printers_asset_bp.route('/supplies/meta', methods=['GET'])
@jwt_required(optional=True)
def supplies_meta():
"""Allowed values for supply type, color, and capacity tier (for the UI)."""
return success_response({
'supplytypes': list(SUPPLY_TYPES),
'colors': list(SUPPLY_COLORS),
'capacitytiers': list(CAPACITY_TIERS),
})
@printers_asset_bp.route('/models', methods=['GET'])
@jwt_required(optional=True)
def list_supply_models():
"""List models with a supply count, for the supply-management picker.
Query parameters:
- search: filter by model number
- vendor_id: filter by vendor
- withsupplies: 'true' to only return models that already have supplies
"""
page, per_page = get_pagination_params(request)
supplycount = db.func.count(ModelSupply.modelsupplyid).label('supplycount')
query = (
db.session.query(Model, Vendor.vendor, supplycount)
.outerjoin(Vendor, Vendor.vendorid == Model.vendorid)
.outerjoin(ModelSupply, db.and_(
ModelSupply.modelnumberid == Model.modelnumberid,
ModelSupply.isactive == True,
))
.group_by(Model.modelnumberid, Vendor.vendor)
)
# Toner/drum/waste only apply to printers, so restrict the picker to
# printer models: those attached to a printer asset, or those that already
# carry supply mappings. Keeps machine/controller models out of the list.
printer_model_ids = (
db.session.query(Printer.modelnumberid)
.filter(Printer.modelnumberid.isnot(None))
)
supply_model_ids = db.session.query(ModelSupply.modelnumberid)
query = query.filter(db.or_(
Model.modelnumberid.in_(printer_model_ids),
Model.modelnumberid.in_(supply_model_ids),
))
if search := request.args.get('search'):
query = query.filter(Model.modelnumber.ilike(f'%{search}%'))
if vendor_id := request.args.get('vendor_id'):
query = query.filter(Model.vendorid == int(vendor_id))
if request.args.get('withsupplies', '').lower() == 'true':
query = query.having(supplycount > 0)
query = query.order_by(Model.modelnumber)
total = query.count()
rows = query.limit(per_page).offset((page - 1) * per_page).all()
data = [{
'modelnumberid': model.modelnumberid,
'modelnumber': model.modelnumber,
'vendor': vendor,
'vendorid': model.vendorid,
'supplycount': count,
} for model, vendor, count in rows]
return paginated_response(data, page, per_page, total)
@printers_asset_bp.route('/models/<int:modelnumberid>/supplies', methods=['GET'])
@jwt_required(optional=True)
def list_model_supplies(modelnumberid: int):
"""List all supplies mapped to a model."""
model = Model.query.get(modelnumberid)
if not model:
return error_response(ErrorCodes.NOT_FOUND, 'Model not found', http_code=404)
supplies = (
ModelSupply.query
.filter_by(modelnumberid=modelnumberid, isactive=True)
.order_by(ModelSupply.supplytype, ModelSupply.color, ModelSupply.capacitytier)
.all()
)
return success_response({
'modelnumberid': modelnumberid,
'modelnumber': model.modelnumber,
'supplies': [s.to_dict() for s in supplies],
})
@printers_asset_bp.route('/models/<int:modelnumberid>/supplies', methods=['POST'])
@jwt_required()
def create_model_supply(modelnumberid: int):
"""Add a supply to a model."""
model = Model.query.get(modelnumberid)
if not model:
return error_response(ErrorCodes.NOT_FOUND, 'Model not found', http_code=404)
data = request.get_json()
message = _validate_supply_payload(data)
if message:
return error_response(ErrorCodes.VALIDATION_ERROR, message)
existing = ModelSupply.query.filter_by(
modelnumberid=modelnumberid,
partnumber=data['partnumber'],
).first()
if existing:
return error_response(
ErrorCodes.CONFLICT,
f"Part number '{data['partnumber']}' already mapped to this model",
http_code=409,
)
supply = ModelSupply(
modelnumberid=modelnumberid,
supplytype=data.get('supplytype', 'toner'),
color=data.get('color', 'none'),
capacitytier=data.get('capacitytier', 'standard'),
partnumber=data['partnumber'],
marketingname=data.get('marketingname'),
pageyield=data.get('pageyield'),
notes=data.get('notes'),
)
db.session.add(supply)
db.session.commit()
return success_response(supply.to_dict(), message='Supply added', http_code=201)
@printers_asset_bp.route('/supplies/<int:modelsupplyid>', methods=['PUT'])
@jwt_required()
def update_model_supply(modelsupplyid: int):
"""Update a model supply."""
supply = ModelSupply.query.get(modelsupplyid)
if not supply:
return error_response(ErrorCodes.NOT_FOUND, 'Supply not found', http_code=404)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
# validate only the fields present
merged = {
'partnumber': data.get('partnumber', supply.partnumber),
'supplytype': data.get('supplytype', supply.supplytype),
'color': data.get('color', supply.color),
'capacitytier': data.get('capacitytier', supply.capacitytier),
}
message = _validate_supply_payload(merged)
if message:
return error_response(ErrorCodes.VALIDATION_ERROR, message)
if 'partnumber' in data and data['partnumber'] != supply.partnumber:
clash = ModelSupply.query.filter_by(
modelnumberid=supply.modelnumberid,
partnumber=data['partnumber'],
).first()
if clash:
return error_response(
ErrorCodes.CONFLICT,
f"Part number '{data['partnumber']}' already mapped to this model",
http_code=409,
)
for field in ('supplytype', 'color', 'capacitytier', 'partnumber',
'marketingname', 'pageyield', 'notes'):
if field in data:
setattr(supply, field, data[field])
db.session.commit()
return success_response(supply.to_dict(), message='Supply updated')
@printers_asset_bp.route('/supplies/<int:modelsupplyid>', methods=['DELETE'])
@jwt_required()
def delete_model_supply(modelsupplyid: int):
"""Delete a model supply."""
supply = ModelSupply.query.get(modelsupplyid)
if not supply:
return error_response(ErrorCodes.NOT_FOUND, 'Supply not found', http_code=404)
db.session.delete(supply)
db.session.commit()
return success_response(message='Supply deleted')

View File

@@ -13,7 +13,7 @@ from shopdb.extensions import db
from shopdb.core.models.machine import MachineType from shopdb.core.models.machine import MachineType
from shopdb.core.models import AssetType from shopdb.core.models import AssetType
from .models import PrinterData, Printer, PrinterType from .models import PrinterData, Printer, PrinterType, ModelSupply
from .api import printers_bp, printers_asset_bp from .api import printers_bp, printers_asset_bp
from .services import ZabbixService from .services import ZabbixService
@@ -77,6 +77,7 @@ class PrintersPlugin(BasePlugin):
PrinterData, # Legacy Machine-based PrinterData, # Legacy Machine-based
Printer, # New Asset-based Printer, # New Asset-based
PrinterType, # New printer type classification PrinterType, # New printer type classification
ModelSupply, # model -> toner/drum/waste part numbers
] ]
def get_services(self) -> Dict[str, Type]: def get_services(self) -> Dict[str, Type]:
@@ -131,6 +132,7 @@ class PrintersPlugin(BasePlugin):
('Laser', 'Standard laser printer', 'printer'), ('Laser', 'Standard laser printer', 'printer'),
('Inkjet', 'Inkjet printer', 'printer'), ('Inkjet', 'Inkjet printer', 'printer'),
('Label', 'Label/barcode printer', 'barcode'), ('Label', 'Label/barcode printer', 'barcode'),
('Card', 'ID / card printer', 'id-card'),
('MFP', 'Multifunction printer with scan/copy/fax', 'printer'), ('MFP', 'Multifunction printer with scan/copy/fax', 'printer'),
('Plotter', 'Large format plotter', 'drafting-compass'), ('Plotter', 'Large format plotter', 'drafting-compass'),
('Thermal', 'Thermal printer', 'temperature-high'), ('Thermal', 'Thermal printer', 'temperature-high'),
@@ -209,6 +211,19 @@ class PrintersPlugin(BasePlugin):
for supply in supplies: for supply in supplies:
click.echo(f" {supply['name']}: {supply['level']}%") click.echo(f" {supply['name']}: {supply['level']}%")
@printerscli.command('seed-supplies')
def seedsuppliescommand():
"""Seed corrected model->toner part numbers into modelsupplies."""
from flask import current_app
from .services import seedsupplies
with current_app.app_context():
summary = seedsupplies()
click.echo(
f"Seeded supplies: {summary['suppliesadded']} added across "
f"{summary['modelstouched']} models."
)
return [printerscli] return [printerscli]
def get_dashboard_widgets(self) -> List[Dict]: def get_dashboard_widgets(self) -> List[Dict]:

View File

@@ -42,7 +42,7 @@ def seed_cli():
def seed_reference_data(): def seed_reference_data():
"""Seed reference data (machine types, statuses, etc.).""" """Seed reference data (machine types, statuses, etc.)."""
from shopdb.extensions import db from shopdb.extensions import db
from shopdb.core.models import MachineType, MachineStatus, OperatingSystem from shopdb.core.models import MachineType, MachineStatus, OperatingSystem, AssetStatus
from shopdb.core.models.relationship import RelationshipType from shopdb.core.models.relationship import RelationshipType
# Machine types # Machine types
@@ -82,6 +82,24 @@ def seed_reference_data():
s = MachineStatus(**s_data) s = MachineStatus(**s_data)
db.session.add(s) db.session.add(s)
# Asset statuses (canonical set - the asset model is the contract)
asset_statuses = [
{'status': 'In Use', 'description': 'Currently in use', 'color': '#28a745'},
{'status': 'Inventory', 'description': 'In inventory', 'color': '#17a2b8'},
{'status': 'In Repair', 'description': 'Being repaired', 'color': '#ffc107'},
{'status': 'Retired', 'description': 'No longer in use', 'color': '#6c757d'},
{'status': 'Returned', 'description': 'Returned to vendor or owner', 'color': '#fd7e14'},
{'status': 'Warrantied', 'description': 'Under warranty service', 'color': '#20c997'},
{'status': 'Lost', 'description': 'Lost or missing', 'color': '#dc3545'},
]
for s_data in asset_statuses:
existing = AssetStatus.query.filter_by(status=s_data['status']).first()
if not existing:
db.session.add(AssetStatus(isactive=True, **s_data))
elif existing.isactive is not True:
existing.isactive = True
# Operating systems # Operating systems
os_list = [ os_list = [
{'osname': 'Windows 10', 'osversion': '10.0'}, {'osname': 'Windows 10', 'osversion': '10.0'},

View File

@@ -53,6 +53,7 @@ class Config:
LOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO') LOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO')
ZABBIX_ENABLED = os.environ.get('ZABBIX_ENABLED', 'false').lower() == 'true'
ZABBIX_URL = os.environ.get('ZABBIX_URL', '') ZABBIX_URL = os.environ.get('ZABBIX_URL', '')
ZABBIX_TOKEN = os.environ.get('ZABBIX_TOKEN', '') ZABBIX_TOKEN = os.environ.get('ZABBIX_TOKEN', '')

View File

@@ -153,6 +153,56 @@ def create_asset_status():
return success_response(s.to_dict(), message='Asset status created', http_code=201) return success_response(s.to_dict(), message='Asset status created', http_code=201)
@assets_bp.route('/statuses/<int:status_id>', methods=['PUT'])
@jwt_required()
def update_asset_status(status_id: int):
"""Update an asset status."""
s = AssetStatus.query.get(status_id)
if not s:
return error_response(ErrorCodes.NOT_FOUND, 'Asset status not found',
http_code=404)
data = request.get_json() or {}
# Conflict check on rename
if 'status' in data and data['status'] != s.status:
if AssetStatus.query.filter_by(status=data['status']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Asset status '{data['status']}' already exists",
http_code=409
)
for key in ('status', 'description', 'color', 'isactive'):
if key in data:
setattr(s, key, data[key])
db.session.commit()
return success_response(s.to_dict(), message='Asset status updated')
@assets_bp.route('/statuses/<int:status_id>', methods=['DELETE'])
@jwt_required()
def delete_asset_status(status_id: int):
"""Delete an asset status. Refused if any asset still uses it."""
s = AssetStatus.query.get(status_id)
if not s:
return error_response(ErrorCodes.NOT_FOUND, 'Asset status not found',
http_code=404)
inuse = Asset.query.filter_by(statusid=status_id).count()
if inuse:
return error_response(
ErrorCodes.CONFLICT,
f"Cannot delete: {inuse} asset(s) still use this status",
http_code=409
)
db.session.delete(s)
db.session.commit()
return success_response(message='Asset status deleted')
# ============================================================================= # =============================================================================
# Assets # Assets
# ============================================================================= # =============================================================================

View File

@@ -35,7 +35,7 @@ PLUGIN_TABLE_OWNERS: dict[str, Iterable[str]] = {
'equipment': ('equipmenttypes', 'equipment'), 'equipment': ('equipmenttypes', 'equipment'),
'network': ('networkdevicetypes', 'networkdevices', 'vlans', 'subnets'), 'network': ('networkdevicetypes', 'networkdevices', 'vlans', 'subnets'),
'notifications': ('notificationtypes', 'notifications'), 'notifications': ('notificationtypes', 'notifications'),
'printers': ('printertypes', 'printers', 'printerdata'), 'printers': ('printertypes', 'printers', 'printerdata', 'modelsupplies'),
'usb': ('usbdevicetypes', 'usbdevices', 'usbcheckouts'), 'usb': ('usbdevicetypes', 'usbdevices', 'usbcheckouts'),
} }