# (c) cavaliba.com - data - schema.py import json import uuid import app_home.cache as cache import yaml from app_data.models import FIELD_FORMAT_CHOICE, DataClass, DataSchema from app_home.log import ERROR, log from django.conf import settings # 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', } 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 # X keyname_label # keyname_mode: edit(*) | auto # displayname_option self.order = 100 self.page = None # _sections # 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_enaled # default_value def __str__(self): return f"{self.classname}" FIELD_DEFAULTS = { 'displayname': '', 'description': '', 'is_enabled': True, 'dataformat_ext': '', 'cardinal_min': 0, 'cardinal_max': 1, 'default_value': '', } @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 # ----------------------------------- # properties # ----------------------------------- # 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") # ----------------------------------- # classmethod # ----------------------------------- @classmethod def exists(cls, classname=None): if classname: return DataClass.objects.filter(keyname=classname).exists() return False @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 # order = field.order # page = field.page # if not page: # page = "Default" # if page not in schema.ordered_fields: # schema.ordered_fields[page] = {} # if order not in schema.ordered_fields[page]: # schema.ordered_fields[page][order]= [] # schema.ordered_fields[page][order].append(field.keyname) # update ordered 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) # ----------------------------------- # method # ----------------------------------- # PERMISSIONS 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 # 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 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 myyaml = {} myyaml["keyname_mode"] = self.keyname_mode myyaml["keyname_label"] = self.keyname_label myyaml["displayname_label"] = self.displayname_label myyaml["icon"] = self.icon # 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() 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'] # VALID_OPTIONS for key, value in datadict['_options'].items(): if key in self._VALID_OPTIONS: if self.options[key] != value: # changed = True self.options[key] = value # 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 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()