"""Custom fields API - definitions per asset type + per-asset values.""" import json import re from flask import Blueprint, request from flask_jwt_extended import jwt_required from shopdb.extensions import db from shopdb.core.models import CustomField, CustomFieldValue, Asset, AssetType from shopdb.core.models.customfield import CUSTOM_FIELD_DATATYPES from shopdb.utils.responses import ( success_response, error_response, ErrorCodes ) from shopdb.utils.authz import require_role customfields_bp = Blueprint('customfields', __name__) def _slugify_key(label): """Build a stable machine key from a label: lowercase, alnum only.""" key = re.sub(r'[^a-z0-9]', '', (label or '').lower()) return key[:50] or 'field' def _normalize_options(raw): """Return a JSON string of a clean list of option strings, or None.""" if not raw: return None if isinstance(raw, str): # Accept newline- or comma-separated text too. parts = re.split(r'[\n,]', raw) elif isinstance(raw, (list, tuple)): parts = raw else: return None cleaned = [str(p).strip() for p in parts if str(p).strip()] return json.dumps(cleaned) if cleaned else None # ============================================================================= # Field definitions # ============================================================================= @customfields_bp.route('', methods=['GET']) @jwt_required(optional=True) def list_fields(): """List custom-field definitions. Filter by ?assettypeid=. ?active=false includes inactive ones.""" query = CustomField.query assettypeid = request.args.get('assettypeid', type=int) if assettypeid: query = query.filter_by(assettypeid=assettypeid) if request.args.get('active', 'true').lower() != 'false': query = query.filter_by(isactive=True) fields = query.order_by(CustomField.sortorder, CustomField.fieldid).all() return success_response([f.to_dict() for f in fields]) @customfields_bp.route('', methods=['POST']) @jwt_required() @require_role('admin') def create_field(): data = request.get_json() or {} assettypeid = data.get('assettypeid') label = (data.get('label') or '').strip() if not assettypeid or not label: return error_response(ErrorCodes.VALIDATION_ERROR, 'assettypeid and label are required') if not db.session.get(AssetType, assettypeid): return error_response(ErrorCodes.VALIDATION_ERROR, 'Unknown assettypeid') datatype = data.get('datatype') or 'text' if datatype not in CUSTOM_FIELD_DATATYPES: return error_response(ErrorCodes.VALIDATION_ERROR, f'datatype must be one of {", ".join(CUSTOM_FIELD_DATATYPES)}') fieldkey = (data.get('fieldkey') or '').strip() or _slugify_key(label) # Reactivate a same-key field that was soft-deleted, instead of colliding. existing = CustomField.query.filter_by(assettypeid=assettypeid, fieldkey=fieldkey).first() if existing: if not existing.isactive: existing.isactive = True existing.label = label existing.datatype = datatype existing.options = _normalize_options(data.get('options')) existing.showondetail = bool(data.get('showondetail', True)) existing.showonform = bool(data.get('showonform', True)) existing.sortorder = data.get('sortorder', 0) db.session.commit() return success_response(existing.to_dict(), message='Reactivated existing field') return error_response(ErrorCodes.CONFLICT, f'A field with key "{fieldkey}" already exists for this asset type') field = CustomField( assettypeid=assettypeid, fieldkey=fieldkey, label=label, datatype=datatype, options=_normalize_options(data.get('options')), showondetail=bool(data.get('showondetail', True)), showonform=bool(data.get('showonform', True)), sortorder=data.get('sortorder', 0), ) db.session.add(field) db.session.commit() return success_response(field.to_dict(), message='Custom field created', http_code=201) @customfields_bp.route('/', methods=['PUT']) @jwt_required() @require_role('admin') def update_field(fieldid): field = db.session.get(CustomField, fieldid) if not field: return error_response(ErrorCodes.NOT_FOUND, 'Custom field not found', http_code=404) data = request.get_json() or {} if 'label' in data: field.label = (data['label'] or '').strip() or field.label if 'datatype' in data: if data['datatype'] not in CUSTOM_FIELD_DATATYPES: return error_response(ErrorCodes.VALIDATION_ERROR, 'invalid datatype') field.datatype = data['datatype'] if 'options' in data: field.options = _normalize_options(data['options']) for key in ('showondetail', 'showonform', 'isactive'): if key in data: setattr(field, key, bool(data[key])) if 'sortorder' in data: field.sortorder = data['sortorder'] db.session.commit() return success_response(field.to_dict(), message='Custom field updated') @customfields_bp.route('/', methods=['DELETE']) @jwt_required() @require_role('admin') def delete_field(fieldid): field = db.session.get(CustomField, fieldid) if not field: return error_response(ErrorCodes.NOT_FOUND, 'Custom field not found', http_code=404) # Drop the field and any stored values for it. CustomFieldValue.query.filter_by(fieldid=fieldid).delete() db.session.delete(field) db.session.commit() return success_response(message='Custom field deleted') # ============================================================================= # Per-asset values # ============================================================================= def _fields_with_values(asset): """Return active field defs for this asset's type, each with the asset's stored value merged in.""" fields = (CustomField.query .filter_by(assettypeid=asset.assettypeid, isactive=True) .order_by(CustomField.sortorder, CustomField.fieldid) .all()) values = {v.fieldid: v.value for v in CustomFieldValue.query.filter_by(assetid=asset.assetid).all()} result = [] for f in fields: item = f.to_dict() item['value'] = values.get(f.fieldid) result.append(item) return result @customfields_bp.route('/asset/', methods=['GET']) @jwt_required(optional=True) def get_asset_fields(assetid): """Active custom fields for an asset's type, merged with its values.""" asset = db.session.get(Asset, assetid) if not asset: return error_response(ErrorCodes.NOT_FOUND, 'Asset not found', http_code=404) return success_response(_fields_with_values(asset)) @customfields_bp.route('/asset/', methods=['PUT']) @jwt_required() @require_role('admin') def save_asset_fields(assetid): """Upsert values for an asset. Body: {values: {fieldid: value, ...}}.""" asset = db.session.get(Asset, assetid) if not asset: return error_response(ErrorCodes.NOT_FOUND, 'Asset not found', http_code=404) data = request.get_json() or {} values = data.get('values') or {} # Only accept fields that belong to this asset's type. valid_ids = {f.fieldid for f in CustomField.query.filter_by(assettypeid=asset.assettypeid).all()} for raw_fieldid, raw_value in values.items(): try: fieldid = int(raw_fieldid) except (ValueError, TypeError): continue if fieldid not in valid_ids: continue value = '' if raw_value is None else str(raw_value) row = CustomFieldValue.query.filter_by(fieldid=fieldid, assetid=assetid).first() if value == '': # Empty clears the stored value. if row: db.session.delete(row) continue if row: row.value = value else: db.session.add(CustomFieldValue(fieldid=fieldid, assetid=assetid, value=value)) db.session.commit() return success_response(_fields_with_values(asset), message='Custom fields saved')