# (c) cavaliba.com - data - data.py import copy import json import yaml from django.conf import settings from django.utils import timezone from django.utils.translation import gettext as _ import app_home.cache as cache from app_data.eav import EavBatch from app_data.models import DataEAV, DataInstance from app_data.related import get_related from app_data.revision import revision_get from app_data.schema import Schema from app_home.configuration import get_configuration from app_home.log import DEBUG, log from .fieldtypes.field_boolean import FieldBoolean from .fieldtypes.field_date import FieldDate from .fieldtypes.field_datetime import FieldDatetime from .fieldtypes.field_enumerate import FieldEnumerate from .fieldtypes.field_external import FieldExternal from .fieldtypes.field_file import FieldFile from .fieldtypes.field_float import FieldFloat from .fieldtypes.field_int import FieldInt from .fieldtypes.field_ipv4 import FieldIPV4 from .fieldtypes.field_password import FieldPassword from .fieldtypes.field_permission import FieldPermission from .fieldtypes.field_schema import FieldSchema from .fieldtypes.field_string import FieldString from .fieldtypes.field_text import FieldText from .fieldtypes.field_time import FieldTime _USER_EXTRA_FIELDS = ("firstname", "lastname", "email", "mobile", "external_id") FIELD_CONSTRUCTOR_TABLE = { "unknown": FieldString, "string": FieldString, "int": FieldInt, "float": FieldFloat, "boolean": FieldBoolean, "ipv4": FieldIPV4, "date": FieldDate, "datetime": FieldDatetime, "time": FieldTime, "text": FieldText, "schema": FieldSchema, "enumerate": FieldEnumerate, "external": FieldExternal, "file": FieldFile, "password": FieldPassword, "permission": FieldPermission, } # YAML export class MyYamlDumper(yaml.SafeDumper): def write_line_break(self, data=None): super().write_line_break(data) if len(self.indents) < 2: super().write_line_break() # ------------------------------------------------------------------------- # bigset/count update # ------------------------------------------------------------------------- def update_bigset(aaa=None): """ set bigset to true in DB schema if many instances ; trigers ajax in UI """ bigset_size = int(get_configuration("data", "DATA_BIGSET_SIZE")) schemas = Schema.listall() for schema in schemas: count = Schema.count_instances(schema.classname) if count > bigset_size: schema.is_bigset = True schema.save() log( DEBUG, aaa=aaa, app="data", view="update_bigset", action="updated", status="OK", data=f"{schema.classname},count={count},bigset={schema.is_bigset} ", ) # print(f"update_bigset: schema={schema.classname} count={count} bigset={schema.is_bigset}") # ------------------------------------------------------------------------- # Instance Helpers # ------------------------------------------------------------------------- # -------------------------------------------------------- # INSTANCE base class # -------------------------------------------------------- class Instance: CLASSNAME = None _SUBCLASS_MAP = { "user": ("app_data.user", "User"), "group": ("app_data.group", "Group"), "role": ("app_data.role", "Role"), } @classmethod def _resolve_class(cls, classname): if classname in cls._SUBCLASS_MAP: import importlib mod_path, cls_name = cls._SUBCLASS_MAP[classname] return getattr(importlib.import_module(mod_path), cls_name) return cls def __init__(self, iobj=None, keyname=None, classname=None, expand=False): self.classname = None # string self.schema = None # Schema() self.keyname = None self.iobj = None self.is_enabled = True # True / False self.displayname = None self.p_read = None self.p_update = None self.p_delete = None self.json = None # self.value = [] self.fields = {} # key = field keyname self.errors = [] self.last_update = None # is_bound => @property # id => @property # NEW EMPTY (or else use from_keyname() classmethod) if not iobj: if classname: self.classname = classname if keyname: self.keyname = keyname # EXISTING else: # check iobj is a DataInstance if type(iobj) is not DataInstance: return # TODO : check iobj is not a new Model self.iobj = iobj self.classname = iobj.classname self.keyname = iobj.keyname self.is_enabled = iobj.is_enabled self.displayname = iobj.displayname self.p_read = iobj.p_read self.p_update = iobj.p_update self.p_delete = iobj.p_delete self.last_update = iobj.last_update try: self.json = json.loads(iobj.data_json) except Exception: self.json = None # create fields, and fill with json if available self.schema = Schema.from_name(self.classname) if self.schema: for fieldname, fieldschema in self.schema.fields.items(): # safety: don't inject a bad external keyname (missing prefix _) if fieldschema["dataformat"] == "external": if not fieldname.startswith("_"): continue constructor = FIELD_CONSTRUCTOR_TABLE.get(fieldschema["dataformat"], FieldString) self.fields[fieldname] = constructor(fieldname, fieldschema, self.json) # expand injected fields if expand: self.expand_injected() # options ... self.set_options() # Done ! @property def id(self): if self.iobj: return self.iobj.id @property def is_bound(self): """if object from DB in self.iobj""" if self.iobj: return True return False # new 3.19 - specific constructors @classmethod def from_keyname(cls, classname=None, keyname=None, expand=False): """existing from DB or None""" classname = classname or getattr(cls, "CLASSNAME", None) if not classname: return if not keyname: return # Recompute (no cache) if expand if not expand: cachekey = f"{classname}::{keyname}" instance = cache.cache2_instance.get(cachekey) # cache hit if instance: return instance # cache miss instobj = DataInstance.objects.filter(classname=classname, keyname=keyname).first() if instobj: derived_class = cls._resolve_class(classname) instance = derived_class(iobj=instobj, expand=expand) instance.cache_set() return instance @classmethod def from_id(cls, id=None, expand=False): """existing from id or None - v3.21.0""" if not id: return # no expand, try cache if not expand: cachekey = f"{id}" instance = cache.cache2_instance.get(cachekey) if instance is not None: derived_class = cls._resolve_class(instance.classname) if isinstance(instance, derived_class): return instance else: print( "!!! instance.from_id with wrong class: ", instance.id, instance.keyname, instance.classname, ) # cache miss / wrong derived_class try: instobj = DataInstance.objects.get(pk=id) except Exception: return derived_class = cls._resolve_class(instobj.classname) instance = derived_class(iobj=instobj, expand=expand) if isinstance(instance, derived_class): instance.cache_set() return instance else: # TODO logger error print( "!!! instance.from_id with wrong class: ", instance.id, instance.keyname, instance.classname, ) # return anyway return instance @classmethod def from_iobj(cls, iobj=None, expand=False): """existing from iobj or None""" if not isinstance(iobj, DataInstance): return # Recompute (no cache) if expand if not expand: try: cachekey = f"{iobj.classname}::{iobj.keyname}" except Exception: return instance = cache.cache2_instance.get(cachekey) if instance: derived_class = cls._resolve_class(iobj.classname) if isinstance(instance, derived_class): return instance # cache miss derived_class = cls._resolve_class(iobj.classname) instance = derived_class(iobj=iobj, expand=expand) if instance: instance.cache_set() return instance @classmethod def get_dn(cls, classname=None, keyname=None): """Return displayname for keyname in this class, or None if not found.""" classname = classname or getattr(cls, "CLASSNAME", None) if not classname or not keyname: return None iobj = ( DataInstance.objects.filter(classname=classname, keyname=keyname) .values("displayname") .first() ) if iobj: return iobj["displayname"] return None @classmethod def get_all_keyname(cls, classname=None): """Return list of all keynames for this class.""" classname = classname or getattr(cls, "CLASSNAME", None) if not classname: return [] return list( DataInstance.objects.filter(classname=classname).values_list("keyname", flat=True) ) @classmethod def count_all(cls, classname=None): classname = classname or getattr(cls, "CLASSNAME", None) if classname: return DataInstance.objects.filter(classname=classname).count() else: return 0 @classmethod def exists(cls, classname=None, keyname=None): classname = classname or getattr(cls, "CLASSNAME", None) if not classname or not keyname: return False return DataInstance.objects.filter(classname=classname, keyname=keyname).exists() # NEXT: iterator @classmethod def iterate_classname(cls, classname=None, first=None, last=None, enabled=None, expand=False): """returns a list[] of Instance()""" classname = classname or getattr(cls, "CLASSNAME", None) if not classname: return [] instances = [] if type(first) is int and type(last) is int: if enabled == "yes": instances = DataInstance.objects.filter(classname=classname, is_enabled=True)[ first:last ] elif enabled == "no": instances = DataInstance.objects.filter(classname=classname, is_enabled=False)[ first:last ] else: instances = DataInstance.objects.filter(classname=classname)[first:last] else: if enabled == "yes": instances = DataInstance.objects.filter(classname=classname, is_enabled=True) elif enabled == "no": instances = DataInstance.objects.filter(classname=classname, is_enabled=False) else: instances = DataInstance.objects.filter(classname=classname) reply = [] for instobj in instances: instance = cls.from_iobj(iobj=instobj, expand=expand) if instance: reply.append(instance) return reply def __str__(self): return f"{self.classname}:{self.keyname}" def print(self): print(f"{self.classname}:{self.keyname}") print(f" id: {self.id}") print(f" is_enabled: {self.is_enabled}") print(f" displayname: {self.displayname}") print(f" is_bound: {self.is_bound}") print(f" is_valid: {self.is_valid()}") print(f" p_read: {self.p_read}") print(f" p_update: {self.p_update}") print(f" p_delete: {self.p_delete}") print(" fields:") # pprint(self.schema) for _nouse, v in self.fields.items(): v.print() # pprint(v.fieldschema) print() def to_json(self, refs=None, rev=None): data = self.get_dict_for_export(refs=refs, rev=rev) data.pop("id", None) return json.dumps(data, indent=4, ensure_ascii=False).encode("utf8") def to_yaml(self, refs=None, rev=None): data = self.get_dict_for_export(refs=refs, rev=rev) data.pop("id", None) return yaml.dump([data], allow_unicode=True, Dumper=MyYamlDumper, sort_keys=False) # default_style='|', # cache_get is in classmethod from_id, from_keyname, ... ; instance doesn't exist yet def cache_set(self): # cache by name if self.classname and self.keyname: cachekey = f"{self.classname}::{self.keyname}" cache.cache2_instance.set(cachekey, self) # cache by id if self.id: cachekey = f"{self.id}" cache.cache2_instance.set(cachekey, self) def cache_purge(self): # by name if self.classname and self.keyname: cachekey = f"{self.classname}::{self.keyname}" cache.cache2_instance.delete(cachekey) # by id if self.id: cachekey = f"{self.id}" cache.cache2_instance.delete(cachekey) # flush specific cache : enumerate cache when an _enumerate instance is modified if self.classname == "_enumerate" and self.keyname: cache.cache2_enumerate.delete(self.keyname) def related(self): return get_related(classname=self.classname, keyname=self.keyname) def ordered_fields(self): def sort_key(item): return self.fields[item].order return sorted(self.fields, key=sort_key) def is_field_true(self, fieldname=None): if not fieldname: return False if fieldname not in self.fields: return False return self.fields[fieldname].is_field_true() def get_attribute_first(self, fieldname): r1 = self.get_attribute(fieldname) try: return r1[0] except Exception: return "" def get_attribute(self, fieldname): """ Returns a List [] of attribute value(s) from instance Convert to obj if FieldSchema or SireneGroup or User ... """ if fieldname == "displayname": return [self.displayname] if fieldname == "keyname": return [self.keyname] if fieldname == "is_enabled": return [self.is_enabled] if fieldname not in self.fields: return return self.fields[fieldname].get_attribute() def set_field_value_single(self, fieldname=None, value=None): if not fieldname: return False if fieldname not in self.fields: return False self.fields[fieldname].value = [value] return True # instance PERMISSIONS def has_valid_aaa(self, aaa=None): if not self.is_bound: return False try: if type(aaa["perms"]) is list: return True except Exception: pass return False def has_read_permission(self, aaa=None): if not self.has_valid_aaa(aaa=aaa): return False # global admin : all schema, all instances, all operations if "p_data_admin" in aaa["perms"]: return True # schema admin : aaa has p_admin on this schema ? (allow direct & stop inheritance) schema = Schema.from_name(self.classname) if schema.has_admin_permission(aaa=aaa): return True # instance read if self.p_read: if self.p_read in aaa["perms"]: return True else: return False # default schema read > all data read return schema.has_read_permission(aaa=aaa) def has_update_permission(self, aaa=None): if not self.has_valid_aaa(aaa=aaa): return False if "p_data_admin" in aaa["perms"]: return True # schema admin (no override) schema = Schema.from_name(self.classname) if schema.has_admin_permission(aaa=aaa): return True # if defined, instance permission rules if self.p_update: if self.p_update in aaa["perms"]: return True else: return False # default to class permission return schema.has_update_permission(aaa=aaa) def has_delete_permission(self, aaa=None): if not self.has_valid_aaa(aaa=aaa): return False if "p_data_admin" in aaa["perms"]: return True # schema admin (no override) schema = Schema.from_name(self.classname) if schema.has_admin_permission(aaa=aaa): return True # if defined, instance permission rules if self.p_delete: if self.p_delete in aaa["perms"]: return True else: return False # default to class permission return schema.has_delete_permission(aaa=aaa) def get_recursive_content(self, fieldname=None, fieldmember=None, fieldrecurse=None, done=None): """get list of Instance() from field "fieldname" , including - self instance - fieldmember *.fieldname - recurse on fieldsubgroup : fieldname, member, sub-subgroups ... """ # Example # ------- # SiteGroup = { # sirene_notify =[ SireneGroup1, 2, ...] # members = [site1, site2] # subgroups = [sitegroup1, sitegroup2 ...] # } # # for sitegroup_obj in message.notify_sitegroup.all(): # instance = Instance(iobj=sitegroup_obj) # xlist = instance.get_recursive_content( # fieldname="notify_group", # fieldmember="members", # fieldrecurse="subgroups", # done=[]) # # done[] contains list of iobj like self if done is None: done = [] reply = [] # recursive: already done ? if self in done: return [] done.append(self) # get field values from self xlist1 = self.get_attribute(fieldname) for z in xlist1: if z: if z not in done: reply.append(z) # reply += xlist1 # get all members from self xlist2 = [] if fieldmember: # if fieldmember in self.schema: if fieldmember in self.fields: xlist2 = self.fields[fieldmember].get_attribute() # get field values from members for item in xlist2: # try to get an instance struct # NOTA: won't work on SireneGroup objects ; use SireneGroups for that purpose if type(item) is DataInstance: instance = Instance(iobj=item) if instance: xlist3 = instance.get_attribute(fieldname) for z in xlist3: if z: if z not in done: reply.append(z) # get all subgroup (fieldrecurse field content) xlist4 = [] if fieldrecurse: if fieldrecurse in self.fields: xlist4 = self.fields[fieldrecurse].get_attribute() # recurse for item in xlist4: if type(item) is DataInstance: if item in done: continue instance = Instance(iobj=item) if instance: xlist5 = instance.get_recursive_content( fieldname=fieldname, fieldmember=fieldmember, fieldrecurse=fieldrecurse, done=done, ) for z in xlist5: if z: if z not in done: reply.append(z) # reply += xlist5 reply = list(set(reply)) return reply # set options # ----------- def set_options(self): # empty instance if not self.schema: return # keyname_mode if self.schema.keyname_mode == "auto": if not self.keyname: self.keyname = Schema.create_keyname() # expand / compute injected fields # -------------------------------- def expand_injected(self): # Refresh all injected / external / enumerate / schema # if new empty Instance, skip update if not self.keyname: return # 1 - remove injected fields (from enumerate / schema) # self.fields will be modified so separate loop inventory and processing remove = [] for fieldname, field in self.fields.items(): if field.is_injected: remove.append(fieldname) for v in remove: self.fields.pop(v) # 2 - resolve external to source # self.fields will be modified so separate loop inventory and processing external = [] for fieldname, field in self.fields.items(): if field.dataformat == "external": external.append(fieldname) # processing for fieldname in external: # field = self.fields[fieldname] # recurse here ; will inject resolved field as schema fields self.inject_resolved_external_field(fieldname) # 3 - schema : inject subfields self.inject_schema_subfields() # 4 - enumerate : inject subfields self.inject_enumerate_subfields() return # ------------------------------------------------- # compute with recurse for a single field # inject computed field in self.fields # ------------------------------------------------- def inject_resolved_external_field(self, fieldname): try: field = self.fields[fieldname] except Exception: return # fieldname must be an external (to a schema) if field.dataformat != "external": return parent_fieldname = field.get_parent_fieldname() if not parent_fieldname: return parent_field = self.fields[parent_fieldname] if parent_field.dataformat != "schema": return remote_classname = parent_field.get_classname() if not remote_classname: return # NEXT : support multi-value try: remote_instance_name = parent_field.value[0] except Exception: return # NEXT : add depth to avoid recurse loops # don't expand (perf) ; will recurse on non expanded field remote_instance = Instance.from_keyname( classname=remote_classname, keyname=remote_instance_name ) if not remote_instance: return if not remote_instance.iobj: return # BUG: if not remote_instance.is_bound: # get remote fieldname remote_fieldname = field.get_remote_fieldname() if not remote_fieldname: return # 2 options here # either remote_fieldname is a schema , or "_"+remote_field is an external if remote_fieldname in remote_instance.fields: # schema remote_fieldname2 = remote_fieldname remote_field = remote_instance.fields[remote_fieldname] elif "_" + remote_fieldname in remote_instance.fields: # external remote_fieldname2 = "_" + remote_fieldname remote_field = remote_instance.fields[remote_fieldname2] else: remote_field = None if not remote_field: return remote_field_dataformat = remote_field.dataformat # remote field is an external => recurse (again) if remote_field_dataformat == "external": remote_instance.inject_resolved_external_field(remote_fieldname2) # access remote resolved field (without prefix _ ) try: # remote_fieldname is without _ newvalue = remote_instance.fields[remote_fieldname].get_value() newclassname = remote_instance.fields[remote_fieldname].get_classname() # bad structure except Exception: return # remote field is a schema => DONE, it's the source elif remote_field_dataformat == "schema": newvalue = remote_field.get_value() newclassname = remote_field.get_classname() # bad structure else: return # inject external schema field w/ name = external fieldname without prefix _ new_fieldname = field.fieldname[1:] new_json = {new_fieldname: newvalue} # get subfields subfields = field.get_subfields() new_dataformat_ext = newclassname + " " + " ".join(subfields) new_schema = { "displayname": field.displayname, "description": new_fieldname, "is_multi": False, "dataformat": "schema", "dataformat_ext": new_dataformat_ext, "cardinal_min": 0, "cardinal_max": 1, "is_injected": True, "injected_type": "external", "page": field.page, "order": field.order, } self.fields[new_fieldname] = FieldSchema(new_fieldname, new_schema, new_json) return def inject_schema_subfields(self): # collect all schema fields to process (2 stage : fields will be altered by injection) schema_fieldnames = [] for fieldname, field in self.fields.items(): if field.dataformat == "schema": schema_fieldnames.append(fieldname) # loop over identified schema fields for fieldname in schema_fieldnames: field = self.fields[fieldname] subfieldnames = field.get_subfields() if not subfieldnames: continue target_classname = field.get_classname() target_instancename = field.get_first_value() # NEXT : multi : field__xx.0, field_xx.1 ... target_instance = Instance.from_keyname( classname=target_classname, keyname=target_instancename ) if not target_instance: continue for fn in subfieldnames: new_fieldname = fieldname + "__" + fn new_displayname = field.displayname + f"({fn})" # try a regular target field: if fn in target_instance.fields: target_field = target_instance.fields[fn] # exclude some dataformat we can't inject if target_field.dataformat in ["schema", "external"]: continue new_dataformat = target_field.dataformat new_schema = { "displayname": new_displayname, "description": new_displayname, "dataformat": new_dataformat, "dataformat_ext": target_field.dataformat_ext, "cardinal_min": target_field.cardinal_min, "cardinal_max": target_field.cardinal_max, "is_injected": True, "injected_type": "schema", "page": field.page, "order": field.order, } try: new_list = target_field.get_value() new_json = {new_fieldname: new_list} except Exception: continue constructor = FIELD_CONSTRUCTOR_TABLE.get(new_dataformat, FieldString) self.fields[new_fieldname] = constructor(new_fieldname, new_schema, new_json) # try a built-in attribute (display_name ...) else: pass # TODO : inject displayname, keyname def inject_enumerate_subfields(self): enumerate = [] for fieldname, field in self.fields.items(): if field.dataformat == "enumerate": enumerate.append(fieldname) for fieldname in enumerate: field = self.fields[fieldname] subfields = field.get_subfields() for k, value in subfields.items(): new_fieldname = fieldname + "__" + k try: (t, v) = value except Exception: continue new_json = {new_fieldname: [v]} new_schema = { "displayname": new_fieldname, "description": new_fieldname, "dataformat": "string", "dataformat_ext": "safe", "cardinal_min": 0, "cardinal_max": 1, "is_injected": True, "injected_type": "enumerate", "page": field.page, "order": field.order, } if t == "int": new_schema["dataformat"] = "int" self.fields[new_fieldname] = FieldInt(new_fieldname, new_schema, new_json) elif t == "boolean": new_schema["dataformat"] = "boolean" self.fields[new_fieldname] = FieldBoolean(new_fieldname, new_schema, new_json) elif t == "float": new_schema["dataformat"] = "float" self.fields[new_fieldname] = FieldFloat(new_fieldname, new_schema, new_json) elif t == "date": new_schema["dataformat"] = "date" self.fields[new_fieldname] = FieldDate(new_fieldname, new_schema, new_json) else: self.fields[new_fieldname] = FieldString(new_fieldname, new_schema, new_json) # NEXT : ipv4, datetime, time # -------- def get_dict_for_ui_form(self): # used for edit in form : new or edit instance_ui = {} # SPECIAL fields instance_ui["id"] = self.id instance_ui["keyname"] = self.keyname instance_ui["displayname"] = self.displayname instance_ui["p_read"] = self.p_read instance_ui["p_update"] = self.p_update instance_ui["p_delete"] = self.p_delete instance_ui["is_enabled"] = self.is_enabled # loop over instance.fields instance_ui["PAGES"] = {} for fieldname in self.ordered_fields(): field = self.fields[fieldname] # don't edit injected fields if field.is_injected: continue # don't edit external fields if field.dataformat == "external": continue if fieldname in self.schema.field_hide_from_edit: continue order = field.order page = field.page if not page: page = self.keyname if page not in instance_ui["PAGES"]: instance_ui["PAGES"][page] = {} if order not in instance_ui["PAGES"][page]: instance_ui["PAGES"][page][order] = [] datapoint = self.fields[fieldname].get_datapoint_ui_edit() instance_ui["PAGES"][page][order].append(datapoint) return instance_ui def get_dict_for_ui_detail(self, skip_external=True, skip_injected=True, skip_enumerate=True): """dict suited for UI list template ; NESTED PAGE and FLAT""" # ui = { # keyname:"xxxx" # displayname:"xxx" # is_enabled:True/False" # p_*3 # First Level / Flat mode # ------------------------ # fieldname: { DATAPOINT } # fieldname: { DATAPOINT } # ... # Nested / Page # ------------- # instance_ui["PAGES"][page][order]= [] # # "PAGES": { # "(page)1": { "(order)100": [ {DATAPOINT}, {}, ... ], # "(order)101": [ {}, {}, ... ], # "(page)2": { }, # ... # } # DATAPOINT = { # datapoint["fieldname"] = self.fieldname # datapoint["displayname"] = self.displayname # datapoint["description"] = self.description # datapoint["dataformat"] = self.dataformat # datapoint["dataformat_ext"] = self.dataformat_ext # datapoint["is_multi"] = self.is_multi() # datapoint["bigset"] = False # datapoint["schema"] = CLASSNAME (for schema field) # datapoint["value"] = '' # } # VALUE : depend on field dataformat (ex. ", ".join(values)") instance_ui = {} # SPECIAL fields P7 instance_ui["id"] = self.id instance_ui["keyname"] = self.keyname instance_ui["displayname"] = self.displayname if not self.displayname: instance_ui["displayname"] = self.keyname instance_ui["p_read"] = self.p_read instance_ui["p_update"] = self.p_update instance_ui["p_delete"] = self.p_delete instance_ui["is_enabled"] = self.is_enabled # loop over instance.fields instance_ui["PAGES"] = {} for fieldname in self.ordered_fields(): field = self.fields[fieldname] if skip_external: if field.dataformat == "external": continue if skip_injected: if field.is_injected: continue if skip_enumerate: if field.injected_type == "enumerate": continue if fieldname in self.schema.field_hide_from_detail: continue order = field.order page = field.page if not page: page = self.keyname if page not in instance_ui["PAGES"]: instance_ui["PAGES"][page] = {} if order not in instance_ui["PAGES"][page]: instance_ui["PAGES"][page][order] = [] datapoint = field.get_datapoint_ui_detail() # nested structure PAGE/ORDER instance_ui["PAGES"][page][order].append(datapoint) # append also at first level instance_ui[fieldname] = copy.deepcopy(datapoint) return instance_ui # alias def to_dict(self): return self.get_dict_for_export() def get_dict_for_export(self, refs=None, rev=None): # { # classname: XX # keyname: XX # displayname:"xxx" # is_enabled:True/False" # "attribname": DATAPOINT, # "attribname": ... # _refs: # user: # group: # schema1: # ... # } # DATAPOINT = value / "value" # DATAPOINT = ["v1", "v2", ...] # refs: list of type keys to inline, e.g. ['user', 'group', 'site'] # '*' expands to all schema names (does not include user/group/role) if refs and "*" in refs: refs = [r for r in refs if r != "*"] + Schema.listall_names() instance_export = {} # SPECIAL fields instance_export["id"] = self.id instance_export["classname"] = self.classname instance_export["keyname"] = self.keyname instance_export["displayname"] = self.displayname if self.p_read: instance_export["p_read"] = self.p_read if self.p_update: instance_export["p_update"] = self.p_update if self.p_delete: instance_export["p_delete"] = self.p_delete instance_export["is_enabled"] = self.is_enabled # REGULAR fields for fieldname in self.ordered_fields(): field = self.fields[fieldname] datapoint = field.get_datapoint_for_export() cardinal, cardinal_min, cardinal_max = field.get_cardinal3() if not (cardinal_min == 0 and cardinal == 0): instance_export[fieldname] = datapoint # Pass 1: collect sets of referenced keys per type buckets = {} if refs: for fieldname in self.ordered_fields(): field = self.fields[fieldname] if not field.value: continue fmt = field.dataformat if fmt == "schema": classname = field.dataformat_ext if classname and classname in refs: buckets.setdefault(classname, set()).update(field.value) # Pass 2: DataEAV queries to fill refs refs = {} for classname, keynames in buckets.items(): # keyname → displayname from any EAV row per keyname dn_map = {} for row in ( DataEAV.objects.filter(classname=classname, keyname__in=keynames) .values("keyname", "displayname") .distinct() ): dn_map.setdefault(row["keyname"], row["displayname"] or "") # extra EAV fields for 'user' extra_map = {} if classname == "user": for row in DataEAV.objects.filter( classname="user", keyname__in=keynames, fieldname__in=_USER_EXTRA_FIELDS ).values("keyname", "fieldname", "value"): extra_map.setdefault(row["keyname"], {})[row["fieldname"]] = row["value"] or "" detail = [] for keyname, displayname in dn_map.items(): entry = {"keyname": keyname, "classname": classname, "displayname": displayname} if classname == "user": for fn in _USER_EXTRA_FIELDS: v = extra_map.get(keyname, {}).get(fn, "") if v: entry[fn] = v detail.append({k: v for k, v in entry.items() if v or k == "keyname"}) if detail: refs[classname] = detail if refs: instance_export["_refs"] = refs if rev: revisions = revision_get(classname=self.classname, keyname=self.keyname, limit=rev) if revisions: instance_export["_revision"] = [ { "login": r.username or "", "date": r.date.strftime("%Y-%m-%dT%H:%M:%S"), "action": r.action or "", } for r in revisions ] return instance_export def get_csv_columns(self): csv_columns = ["classname", "keyname", "displayname", "is_enabled"] # ,"p_read","p_update","p_delete for fieldname in self.ordered_fields(): field = self.fields[fieldname] # if field.is_multi(): # continue if field.is_injected: continue if fieldname.startswith("_"): continue if field.dataformat == "text": continue csv_columns.append(fieldname) return csv_columns def get_csv_line(self, csv_columns): # /!\ must match get_csv_columns line = [self.classname, self.keyname, self.displayname, self.is_enabled] # self.p_read, self.p_update,self.p_delete for fieldname in self.ordered_fields(): if fieldname not in csv_columns: continue line.append(self.fields[fieldname].get_csv_cell()) return line # new 3.20 : merged new & edit method def merge_request(self, request, aaa=None): # keyname : editable only for new objects (PK in DB) if not self.is_bound: if self.schema.keyname_mode == "edit": self.keyname = request.POST.get("keyname", "") # displayname (check in is_valid + template auto escaping) self.displayname = request.POST.get("displayname", "") # is_enabled if "is_enabled" in request.POST: self.is_enabled = request.POST["is_enabled"] in settings.TRUE_LIST else: self.is_enabled = False # permissions if type(aaa) is dict: if "perms" in aaa: if "p_data_security_edit" in aaa["perms"]: if "p_read" in request.POST: self.p_read = request.POST["p_read"] if "p_update" in request.POST: self.p_update = request.POST["p_update"] if "p_delete" in request.POST: self.p_delete = request.POST["p_delete"] for fieldname, field in self.fields.items(): # don't merge injected fields if field.is_injected: continue # don't merge external fields if field.dataformat == "external": continue self.fields[fieldname].merge_request(request) # injected fields self.expand_injected() # options self.set_options() def merge_import(self, data, aaa=None): # keyname if new instance only, and not auto computed keyname if not self.is_bound: if self.schema.keyname_mode == "edit": # keyname may be absent # (new Instance with initial keyname, merge additional data w/o keyname) if "keyname" in data: self.keyname = data.get("keyname", None) # displayname if "displayname" in data: self.displayname = data["displayname"] # is_enabled if "is_enabled" in data: self.is_enabled = data["is_enabled"] in settings.TRUE_LIST # else: # self.is_enabled = False # permissions if type(aaa) is dict: if "perms" in aaa: if "p_data_security_edit" in aaa["perms"]: if "p_read" in data: self.p_read = data["p_read"] if "p_update" in data: self.p_update = data["p_update"] if "p_delete" in data: self.p_delete = data["p_delete"] for fieldname, field in self.fields.items(): # don't merge injected fields if field.is_injected: continue # don't merge external fields if field.dataformat == "external": continue # special case : Boolean are not in POST request if False if fieldname in data: fielddata = data[fieldname] self.fields[fieldname].merge_import(fielddata) # injected fields self.expand_injected() # options self.set_options() def is_valid(self): reply = True # keyname if not self.keyname: self.errors.append(_("missing keyname")) return False if len(self.keyname) == 0: self.errors.append(_("keyname empty")) return False # classname (schema) if not self.classname: self.errors.append(_("missing classname")) return False if len(self.classname) == 0: self.errors.append(_("empty classname")) return False schema = Schema.from_name(classname=self.classname) if not schema: self.errors.append(_("unknown classname")) return False # NEXT: displayname = safe string for display ? # is_enabled if type(self.is_enabled) is not bool: return False # NEXT - check permissions ? for _fieldname, field in self.fields.items(): r = field.is_valid() if not r: reply = False # err = fieldname err = field.displayname self.errors.append(str(err)) # NEXT: use field.errors info return reply # ACTIONs : update Instance() and iobj ; call save to DB # ------------------------------------------------------- def enable(self): if not self.is_bound: return False self.iobj.is_enabled = True self.is_enabled = True return self.save() def disable(self): if not self.is_bound: return False self.is_enabled = False self.iobj.is_enabled = False return self.save() def init(self): """init only if not existing""" # already exists in struct? if self.iobj: self.errors.append(_("init - already created - can't recreate")) return False # instance already in DB ? if Instance.exists(classname=self.classname, keyname=self.keyname): self.errors.append(_("init - instance already in DB - can't recreate")) return False # create new instance self.iobj = DataInstance() self.iobj.classname = self.classname return self.update() def create(self): """Create or update new Instance in DB from 'instance'""" # compute/update keyname, ... self.set_options() # create if doesn't exist if not self.is_bound: iobj = DataInstance() iobj.classname = self.classname # new V3.19 iobj.keyname = self.keyname iobj.is_enabled = self.is_enabled iobj.p_read = self.p_read iobj.p_update = self.p_update iobj.p_delete = self.p_delete # iobj.save() self.iobj = iobj return self.update() def update(self): """ update self.iobj with Instance struct , and save to DB """ if not self.is_bound: return False self.set_options() self.iobj.keyname = self.keyname self.iobj.displayname = self.displayname self.iobj.is_enabled = self.is_enabled self.iobj.p_read = self.p_read self.iobj.p_update = self.p_update self.iobj.p_delete = self.p_delete # V3.19 - populate new classname field self.iobj.classname = self.classname data = {} for fieldname, field in self.fields.items(): if field.is_injected: continue field_datalist = self.fields[fieldname].get_value() # don't store empty fields in DB if len(field_datalist) == 0: continue # external field : restore fieldname without _ prefix if field.dataformat == "external": original_fieldname = fieldname[1:] data[original_fieldname] = field_datalist else: data[fieldname] = field_datalist # NEXT : strict mode : self.is_valid() must be True try: self.iobj.data_json = json.dumps(data, indent=2, ensure_ascii=False) except Exception: print("Invalid JSON in update()") return False return self.save() # ----------------- # real write to DB # ----------------- def delete(self): """ Place #1 for real write to DB : delete self.iobj from DB (instance + EAV) """ if not self.is_bound: return False self.cache_purge() try: self.iobj.delete() except Exception as e: print(f"DB delete failed for {self.classname}:{self.keyname} - {e}") return False self.iobj = None self.last_update = timezone.now() EavBatch.save(self) return True def save(self): """ Place #2 for real write to DB : write self.iobj to DB (instance + EAV) WARNING: doesnt merge User() structure to iobj ; done in update() """ if not self.iobj: print("!! no iobj in save()") return False # populate new classname field self.iobj.classname = self.classname self.last_update = timezone.now() self.iobj.last_update = self.last_update self.cache_purge() try: self.iobj.save() except Exception as e: print(f"save() to DB failed: {e}") return False EavBatch.save(self) return True