# (c) cavaliba.com - data - schema.py import json import uuid import yaml from django.conf import settings import app_home.cache as cache from app_data.models import FIELD_FORMAT_CHOICE, DataClass, DataInstance, DataSchema from app_home.log import ERROR, log # New V3.19 class Schema: _RESERVED = [ "classname", "keyname", "displayname", "is_enabled", "order", "page", "_action", "_options", "_schema", ] _VALID_OPTIONS = { "icon": "fa-question", "keyname_mode": "edit", "keyname_label": "Key", "displayname_label": "Name", "notify": [], "field_hide_from_detail": [], "field_hide_from_edit": [], } FIELD_DEFAULTS = { "displayname": "", "description": "", "is_enabled": True, "dataformat_ext": "", "cardinal_min": 0, "cardinal_max": 1, "default_value": "", } def __init__(self): self.classname = None self.displayname = None self.is_enabled = True self.options = {} for k, v in self._VALID_OPTIONS.items(): self.options[k] = v # @properties # X icon # X keyname_mode : edit(*) | auto # X keyname_label self.order = 100 self.page = None # permissions self.p_admin = None self.p_create = None self.p_read = None self.p_update = None self.p_delete = None # other # NO: self.count_estimation = 0 self.is_bigset = False # fields self.ordered_fields = {} # self.ordered_fields['page'][order] = fieldname self.fields = {} # fieldname => dict {} # displayname # description # dataformat # dataformat_ext # order # page # cardinal_min # cardinal_max # is_enabled # default_value def __str__(self): return f"{self.classname}" @staticmethod def get_empty_field_dict(): return { "displayname": "", "description": "", "is_enabled": True, "dataformat": "string", "dataformat_ext": "", "order": 100, "page": "Default", "cardinal_min": 0, "cardinal_max": 1, "default_value": "", } @staticmethod def update_field_from_dict(fielddict, fielddatadict): """ returns: True : fielddict has changed False: no change """ changed = False for k, v in fielddatadict.items(): if k not in fielddict: continue new_value = fielddatadict[k] if k == "dataformat" and new_value not in dict(FIELD_FORMAT_CHOICE): log( ERROR, app="data", view="schema", action="update_field_from_dict", status="KO", data=f"invalid dataformat '{new_value}' for field '{fielddict.get('keyname', '?')}' - fallback to unknown", ) new_value = "unknown" if fielddict[k] != new_value: fielddict[k] = new_value changed = True return changed # icon - V3.20 # ------------ @property def icon(self): return self.options.get("icon") @icon.setter def icon(self, value): # self.icon = value if value: self.options["icon"] = value else: self.options["icon"] = "fa-question" # keyname_mode: edit(*)|auto - V3.19 # ---------------------------------- @property def keyname_mode(self): v = self.options.get("keyname_mode") if v not in ["edit", "auto"]: v = "edit" return v @keyname_mode.setter def keyname_mode(self, value): if value in ["edit", "auto"]: self.options["keyname_mode"] = value else: self.options["keyname_mode"] = "edit" @classmethod def create_keyname(self): return str(uuid.uuid4()) # keyname_label - V3.20 # ---------------------- @property def keyname_label(self): return self.options.get("keyname_label", "Keyname") # displayname_label - V3.20 # --------------------------- @property def displayname_label(self): return self.options.get("displayname_label", "Displayname") # notify - v4.0 # --------------- # notify: # - template: app_maintenance # label: "Notify App Maintenance" # - template: app_incident # label: "Notify App Incident" # - template: _template # label: "Notify from template..." # - template: ~ # YAML null → Python None → blank message # label: "Notify from blank message" # - template: _self # sirene_template Schema only # label: "Create message" @property def notify(self): if hasattr(self, "_notify_resolved"): return self._notify_resolved raw = self.options.get("notify", []) if not isinstance(raw, list): self._notify_resolved = {} return self._notify_resolved result = {} for item in raw: if not isinstance(item, dict): continue label = item.get("label", "?") template_key = item.get("template") if template_key in (None, "_self", "_template"): result[label] = template_key else: tid = ( DataInstance.objects.filter( classname="sirene_template", keyname=template_key, is_enabled=True, ) .values_list("id", flat=True) .first() ) if tid is not None: result[label] = tid self._notify_resolved = result return result @property def field_hide_from_detail(self): v = self.options.get("field_hide_from_detail", []) if not isinstance(v, list): return [] return [x for x in v if isinstance(x, str)] @property def field_hide_from_edit(self): v = self.options.get("field_hide_from_edit", []) if not isinstance(v, list): return [] return [x for x in v if isinstance(x, str)] # ----------------------------------- # classmethod # ----------------------------------- @classmethod def exists(cls, classname=None): if classname: return DataClass.objects.filter(keyname=classname).exists() return False @classmethod def count_instances(cls, classname=None): if not classname: return 0 return DataInstance.objects.filter(classname=classname).count() @classmethod def listall_obj(cls): """returns a list[] of all DataClass as Db Obj""" return DataClass.objects.all() @classmethod def listall(cls): """returns a list[] of all DataClass as Schema() objects""" reply = [] names = DataClass.objects.values_list("keyname", flat=True).order_by("order").all() for name in names: schema = cls.from_name(name) reply.append(schema) return reply @classmethod def listall_names(cls): """returns a list[str->] of all DataClass (Schema) names""" return list(DataClass.objects.values_list("keyname", flat=True).all()) @classmethod def displayname_dict(cls): """returns a dict of all classname => displayname""" # [ { keyname:displayname}, {}, ...] names_list = DataClass.objects.values("keyname", "displayname").all() # to pure dict : { keyname: displayname} names = {} for adict in names_list: names[adict["keyname"]] = adict["displayname"] return names @classmethod def from_obj(cls, obj=None): """load from DB obj""" if not obj: return try: classname = obj.keyname except Exception: return schema = cls() schema.classname = classname schema.displayname = obj.displayname schema.is_enabled = obj.is_enabled # first level options (3.20) >> @property # schema.icon = obj.icon schema.page = obj.page schema.order = obj.order schema.is_bigset = obj.is_bigset # NO: schema.count_estimation = obj.count_estimation schema.p_admin = obj.p_admin schema.p_create = obj.p_create schema.p_read = obj.p_read schema.p_update = obj.p_update schema.p_delete = obj.p_delete # options try: schema.options = yaml.safe_load(obj.options) if type(schema.options) is not dict: schema.options = {} except Exception: schema.options = {} db_fields = DataSchema.objects.filter(classname=classname).order_by("order") schema.fields = {} # schema.ordered_fields = {} # self.ordered_fields['page'][order] = [fieldname, ...] if db_fields: for field in db_fields: m = {} m = Schema.get_empty_field_dict() if field.displayname and len(field.displayname) > 0: m["displayname"] = field.displayname if len(field.description) > 0: m["description"] = field.description m["dataformat"] = field.dataformat if field.dataformat_ext and len(field.dataformat_ext) > 0: m["dataformat_ext"] = field.dataformat_ext m["order"] = field.order if field.page and len(field.page) > 0: m["page"] = field.page if field.cardinal_min > 0: m["cardinal_min"] = field.cardinal_min if field.cardinal_max != 1: m["cardinal_max"] = field.cardinal_max if not field.is_enabled: m["is_enabled"] = field.is_enabled if field.default_value and len(field.default_value) > 0: m["default_value"] = field.default_value schema.fields[field.keyname] = m schema.update_ordered() return schema @classmethod def from_id(cls, id=None): """existing from id or None - v3.32.0 - no caching""" if not id: return try: obj = DataClass.objects.get(pk=id) except Exception: return schema = Schema.from_obj(obj) if schema: cache.cache2_schema.set(schema.classname, schema) return schema @classmethod def from_name(cls, classname=None): """load from cache or DB by classname or None""" if not classname: return # warning: cachekey is also know by delete() methods a = cache.cache2_schema.get(classname) if a: return a obj = DataClass.objects.filter(keyname=classname).first() if not obj: return schema = Schema.from_obj(obj) if schema: cache.cache2_schema.set(classname, schema) return schema def update_ordered(self): self.ordered_fields = {} for fieldname, fielddict in self.fields.items(): order = fielddict["order"] page = fielddict["page"] if not page: page = "Default" if page not in self.ordered_fields: self.ordered_fields[page] = {} if order not in self.ordered_fields[page]: self.ordered_fields[page][order] = [] self.ordered_fields[page][order].append(fieldname) # UT @classmethod def delete(cls, classname=None): # delete schema (DataClass) schema_obj = DataClass.objects.filter(keyname=classname).first() if schema_obj: schema_obj.delete() cache.cache2_schema.delete(classname) cls.purge_instance_cache(classname) # Delete fields (DataSchema) DataSchema.objects.filter(classname=classname).delete() return schema_obj # UT @classmethod def enable(cls, classname=None): schema_obj = DataClass.objects.filter(keyname=classname).first() if schema_obj: schema_obj.is_enabled = True schema_obj.save() cache.cache2_schema.delete(classname) cls.purge_instance_cache(classname) return schema_obj @classmethod def disable(cls, classname=None): schema_obj = DataClass.objects.filter(keyname=classname).first() if schema_obj: schema_obj.is_enabled = False schema_obj.save() cache.cache2_schema.delete(classname) cls.purge_instance_cache(classname) return schema_obj @classmethod def purge_instance_cache(cls, classname=None): """Purge all cache2_instance entries for the given classname.""" if not classname: return cache.cache2_instance.delete_by_prefix(classname) def has_valid_aaa(self, aaa=None): try: if type(aaa["perms"]) is list: return True except Exception: pass return False def has_global_perms(self, aaa=None): # global p_data_admin if "p_data_admin" in aaa["perms"]: return True # p_admin on self if self.p_admin: if self.p_admin in aaa["perms"]: return True return False def has_admin_permission(self, aaa=None): """check p_admin permission on objects of schema SELF""" if not self.has_valid_aaa(aaa=aaa): return False return self.has_global_perms(aaa=aaa) def has_read_permission(self, aaa=None): """check permission to read SELF objects""" if not self.has_valid_aaa(aaa=aaa): return False if self.has_global_perms(aaa=aaa): return True # p_read on single class rules if self.p_read: if self.p_read in aaa["perms"]: return True else: return False # default to all-class permission if "p_data_read" in aaa["perms"]: return True def has_create_permission(self, aaa=None): """check permission to create SELF objects""" if not self.has_valid_aaa(aaa=aaa): return False if self.has_global_perms(aaa=aaa): return True # p_create on this schema only if self.p_create: if self.p_create in aaa["perms"]: return True else: return False # default to all-class permission if "p_data_create" in aaa["perms"]: return True def has_update_permission(self, aaa=None): """check permission to update SELF objects""" if not self.has_valid_aaa(aaa=aaa): return False if self.has_global_perms(aaa=aaa): return True # p_update on this schema only if self.p_update: if self.p_update in aaa["perms"]: return True else: return False # default to all-class permission if "p_data_update" in aaa["perms"]: return True def has_delete_permission(self, aaa=None): """check permission to update SELF objects""" if not self.has_valid_aaa(aaa=aaa): return False if self.has_global_perms(aaa=aaa): return True # p_delete on this schema only if self.p_delete: if self.p_delete in aaa["perms"]: return True else: return False # default to all-class permission if "p_data_delete" in aaa["perms"]: return True # ------- def to_dict(self): reply = {} reply["classname"] = "_schema" reply["keyname"] = self.classname if self.displayname and len(self.displayname) > 0: reply["displayname"] = self.displayname if self.is_enabled: reply["is_enabled"] = True else: reply["is_enabled"] = False reply["_options"] = {} if self.icon and len(self.icon) > 0: reply["_options"]["icon"] = self.icon reply["_options"]["keyname_mode"] = self.keyname_mode if self.keyname_label and len(self.keyname_label) > 0: reply["_options"]["keyname_label"] = self.keyname_label if self.displayname_label and len(self.displayname_label) > 0: reply["_options"]["displayname_label"] = self.displayname_label if self.p_admin and len(self.p_admin) > 0: reply["_options"]["p_admin"] = self.p_admin if self.p_create and len(self.p_create) > 0: reply["_options"]["p_create"] = self.p_create if self.p_read and len(self.p_read) > 0: reply["_options"]["p_read"] = self.p_read if self.p_update and len(self.p_update) > 0: reply["_options"]["p_update"] = self.p_update if self.p_delete and len(self.p_delete) > 0: reply["_options"]["p_delete"] = self.p_delete # v4.0 raw_notify = self.options.get("notify", []) if raw_notify: reply["_options"]["notify"] = raw_notify raw_hide = self.options.get("field_hide_from_detail", []) if raw_hide: reply["_options"]["field_hide_from_detail"] = raw_hide raw_hide_edit = self.options.get("field_hide_from_edit", []) if raw_hide_edit: reply["_options"]["field_hide_from_edit"] = raw_hide_edit # NEXT: sections # fields # self.ordered_fields['page'][order] = [fieldname, ] for page in self.ordered_fields: for order in self.ordered_fields[page]: for keyname in self.ordered_fields[page][order]: field = self.fields[keyname] reply[keyname] = { k: v for k, v in field.items() if k not in Schema.FIELD_DEFAULTS or v != Schema.FIELD_DEFAULTS[k] } return reply def to_yaml(self): content = [self.to_dict()] return yaml.dump(content, allow_unicode=True, sort_keys=False, default_flow_style=False) def to_json(self): content = [self.to_dict()] return json.dumps(content, indent=2, ensure_ascii=False).encode("utf8") def update_from_dict(self, datadict, verbose=False): """ merge (yaml) dict into existing schema (may be new/unsaved) """ # keyname (used when loading _schema objects) if "keyname" in datadict: if self.classname != datadict["keyname"]: self.classname = datadict["keyname"] # displayname if "displayname" in datadict: if self.displayname != datadict["displayname"]: # changed = true => DataRevision self.displayname = datadict["displayname"] # is_enabled if "is_enabled" in datadict: value = datadict["is_enabled"] if isinstance(value, str): value = value.lower() in settings.TRUE_LIST if self.is_enabled != value: # changed = True self.is_enabled = value # page if "page" in datadict: if self.page != datadict["page"]: # changed = True self.page = datadict["page"] # order if "order" in datadict: order_val = datadict["order"] if isinstance(order_val, str): try: order_val = int(order_val) except (ValueError, TypeError): order_val = None if self.order != order_val: # changed = True self.order = order_val if "_options" in datadict: if isinstance(datadict["_options"], dict): # permissions # stored in self. ; stored as attrib in DB # BUT in YAML, inside _options (to avoid fieldname collision) if "p_admin" in datadict["_options"]: if self.p_admin != datadict["_options"]["p_admin"]: # changed = True self.p_admin = datadict["_options"]["p_admin"] if "p_create" in datadict["_options"]: if self.p_create != datadict["_options"]["p_create"]: # changed = True self.p_create = datadict["_options"]["p_create"] if "p_update" in datadict["_options"]: if self.p_update != datadict["_options"]["p_update"]: # changed = True self.p_update = datadict["_options"]["p_update"] if "p_delete" in datadict["_options"]: if self.p_delete != datadict["_options"]["p_delete"]: # changed = True self.p_delete = datadict["_options"]["p_delete"] if "p_read" in datadict["_options"]: if self.p_read != datadict["_options"]["p_read"]: # changed = True self.p_read = datadict["_options"]["p_read"] # FILTER VALID_OPTIONS for key, value in datadict["_options"].items(): if key in self._VALID_OPTIONS: self.options[key] = value # if self.options[key] != value: # # changed = True # self.options[key] = value # notify - V4.0 (list of dicts, handled before generic loop) if key == "notify": self.options[key] = value if isinstance(value, list) else [] if key == "field_hide_from_detail": if isinstance(value, list): self.options[key] = [str(v) for v in value if isinstance(v, str)] else: self.options[key] = [] if key == "field_hide_from_edit": if isinstance(value, list): self.options[key] = [str(v) for v in value if isinstance(v, str)] else: self.options[key] = [] # Fields for fieldname, fielddata in datadict.items(): if fieldname in self._RESERVED: continue # if fieldname.startswith('_'): # continue # Skip non-dict entries (should already be filtered by _RESERVED check) if not isinstance(fielddata, dict): continue # v4.0 - backward compatibility for user, group, role # convert to schema/xxx without _ dataformat = fielddata.get("dataformat", "?") dataformat_ext = fielddata.get("dataformat_ext", "?") if dataformat in ["user", "group", "role"]: fielddata["dataformat"] = "schema" fielddata["dataformat_ext"] = dataformat if dataformat in ["_user", "_group", "_role"]: fielddata["dataformat"] = "schema" fielddata["dataformat_ext"] = dataformat[1:] if dataformat == "schema" and dataformat_ext in ["_user", "_group", "_role"]: dataformat_ext = dataformat_ext[1:] fielddata["dataformat_ext"] = dataformat_ext field_action = fielddata.get("_action", "create") # field delete if field_action == "delete": # delete from Schema if fieldname in self.fields: self.fields.pop(fieldname) # changed # delete operation in DB is performed in self.save() method elif field_action == "enable": if fieldname in self.fields: self.fields[fieldname]["is_enabled"] = True # changed elif field_action == "disable": if fieldname in self.fields: self.fields[fieldname]["is_enabled"] = False # changed elif field_action == "init": if fieldname in self.fields: continue self.fields[fieldname] = Schema.get_empty_field_dict() _ = Schema.update_field_from_dict(self.fields[fieldname], fielddata) # changed = r elif field_action == "create": if fieldname not in self.fields: self.fields[fieldname] = Schema.get_empty_field_dict() _ = Schema.update_field_from_dict(self.fields[fieldname], fielddata) # changed = r elif field_action == "update": if fieldname not in self.fields: continue _ = Schema.update_field_from_dict(self.fields[fieldname], fielddata) # changed = r else: pass # update ordered_field self.update_ordered() def save(self): """save self to DataClass DB""" # exists ? if not self.classname: return obj = DataClass.objects.filter(keyname=self.classname).first() if not obj: obj = DataClass() obj.keyname = self.classname obj.displayname = self.displayname obj.is_enabled = self.is_enabled # obj.icon = self.icon obj.page = self.page obj.order = self.order # options myyaml = {} myyaml["keyname_mode"] = self.keyname_mode myyaml["keyname_label"] = self.keyname_label myyaml["displayname_label"] = self.displayname_label myyaml["icon"] = self.icon # v4.0 myyaml["notify"] = self.options.get("notify", []) myyaml["field_hide_from_detail"] = self.options.get("field_hide_from_detail", []) myyaml["field_hide_from_edit"] = self.options.get("field_hide_from_edit", []) # NEXT : duplicate first level options (icon, p_*) in options field ? obj.options = yaml.dump(myyaml, allow_unicode=True, sort_keys=True) # permissions remain first level attributes # although under '_options:' in external YAML files obj.p_admin = self.p_admin obj.p_create = self.p_create obj.p_read = self.p_read obj.p_update = self.p_update obj.p_delete = self.p_delete # update fields in DataSchema Model # 1- save Schema Fields to DB for fieldname, fielddata in self.fields.items(): fieldobj = DataSchema.objects.filter( classname=self.classname, keyname=fieldname ).first() if not fieldobj: fieldobj = DataSchema() fieldobj.classname = self.classname fieldobj.keyname = fieldname # update provided params if provided only ! if "displayname" in fielddata: fieldobj.displayname = fielddata.get("displayname") if "description" in fielddata: fieldobj.description = fielddata.get("description") if "is_enabled" in fielddata: fieldobj.is_enabled = fielddata.get("is_enabled") in settings.TRUE_LIST if "order" in fielddata: fieldobj.order = int(fielddata.get("order")) if "page" in fielddata: fieldobj.page = fielddata.get("page") if "dataformat" in fielddata: fieldobj.dataformat = fielddata.get("dataformat") if "dataformat_ext" in fielddata: fieldobj.dataformat_ext = fielddata.get("dataformat_ext") if "default_value" in fielddata: fieldobj.default_value = fielddata.get("default_value") if "cardinal_min" in fielddata: fieldobj.cardinal_min = int(fielddata.get("cardinal_min")) if "cardinal_max" in fielddata: fieldobj.cardinal_max = int(fielddata.get("cardinal_max")) fieldobj.save() # 2. Delete orphan Fields from DB fieldobjs = DataSchema.objects.filter(classname=self.classname).all() for fieldobj in fieldobjs: if fieldobj.keyname not in self.fields: fieldobj.delete() # delete from cache cache.cache2_schema.delete(self.classname) self.purge_instance_cache(self.classname) obj.save()