# c(c) cavaliba.com - data - dataview.py """ DataView: named column layout for a schema's list UI ("_dataview" Instances), resolving which columns to show and how to extract/format each Instance's cell values. dataview_for_request() picks the active view (request > session > single-choice > CLASSNAME_default > built-in default) for a given classname. Method inventory: Module functions: get_dataview_dict, get_dataview_content_by_name, dataview_for_request DataView: __init__, get_columns, filter, print """ import re import yaml from django.utils.translation import gettext as _ from app_data.data import Instance, field_export_match from app_data.models import DataEAV # columns not backed by a schema field, always kept regardless of the fields filter # (matches id/classname/keyname being unaffected everywhere else - see # field_export_match()). displayname/last_update/last_sync are deliberately NOT # here even though they're builtin attributes too: they route through # field_export_match() like a regular field, so an explicit "!displayname" (etc.) # filter can exclude them. _DATAVIEW_BUILTIN_SOURCES = {"keyname", "classname", "is_enabled"} # --------------------------------------------------------------------- # Helper # --------------------------------------------------------------------- def get_dataview_dict(schemaname): """ get available dataviews for schema use DataEAV cache IN: schema name (string OUT: dict { keyname:displayname , ... } """ reply = {} eav_entries = DataEAV.objects.filter( classname="_dataview", fieldname="target_class", value=schemaname ).values("keyname", "displayname") # session > single > CLASS_default > DEFAULT built-in dataview_name = None dv_found = False dataview_default = classname + "_default" dataview_selector = get_dataview_dict(classname) # request dataview_name = request.GET.get("dv", None) if not dataview_name: dataview_name = request.POST.get("dv", None) if dataview_name: if dataview_name == dataview_default: dv_found = True else: m = re.compile(r"[a-zA-Z0-9()_/.-]*$") if m.match(dataview_name): if dataview_name in dataview_selector: dv_found = True # in session ? if not dv_found: try: dataview_name = request.session["dataviews"][classname] if dataview_name: m = re.compile(r"[a-zA-Z0-9()_/.-]*$") if m.match(dataview_name): if dataview_name in dataview_selector: dv_found = True except Exception: pass # single entry in selector ? if not dv_found: if len(dataview_selector) == 1: # dataview_name = dataview_selector[0] dataview_name = next(iter(dataview_selector)) dv_found = True # CLASSNAME_default if not dv_found: if dataview_default in dataview_selector: dataview_name = dataview_default dv_found = True # get or built-in default if dv_found: dataview = DataView(keyname=dataview_name) else: dataview = DataView() if dataview.target_class != classname: # wrong class, use a default view dataview = DataView() # store dataview in user session if update_session: if dataview_name: if "dataviews" not in request.session: request.session["dataviews"] = {classname: dataview_name} # request.session.modified = True else: request.session["dataviews"][classname] = dataview_name request.session.modified = True # store request specifics in dataview object if dataview_default not in dataview_selector: dataview.default = dataview_default # dataview.selector = {} dataview.selector = dataview_selector # for k,v in dataview_selector.items(): # #dataview.selector[k] = v.displayname # dataview.selector[k] = v return dataview # ------------------------------------------- # DataView CLASS # ------------------------------------------- class DataView: def __init__(self, keyname=None): """Build the built-in default view (keyname/displayname/last_update columns) if keyname is None or not found/bound; otherwise load and parse the "_dataview" Instance's target_class/content/columns.""" self.instance_id = None self.target_class = None self.keyname = keyname self.displayname = _("Default View") self.is_enabled = True self.content = {"columns": ["keyname", "displayname", "last_update"]} self.columns = ["keyname", "displayname", "last_update"] # request specifics self.default = None # name of default if not in selector self.selector = None # list of available dataviews for target_class if not keyname: return # query _dataview Schema for keyname instance = Instance.from_keyname(classname="_dataview", keyname=keyname) if not instance: return if not instance.is_bound: return self.instance_id = instance.id try: self.target_class = instance.fields["target_class"].value[0] except Exception: self.target_class = None self.displayname = instance.displayname self.is_enabled = instance.is_enabled try: raw = instance.fields["content"].value[0] except Exception: raw = "" try: self.content = yaml.safe_load(raw) except Exception as e: print(f"ERR - can't parse dataview content ({keyname}): {e}") return # Extract columns # columns: # - nice name: # from: keyname # - col1 # - site: # from: mysitename_col2 # yaml_columns = self.content.get("columns", None) computed_columns = [] if yaml_columns: for head in yaml_columns: if type(head) is str: computed_columns.append(head) elif type(head) is dict: for col, _nouse in head.items(): computed_columns.append(col) self.columns = computed_columns # NEXT: if no keyname/displayname , add keyname ? def _column_source(self, head): """Return the underlying instance field name for a raw column head entry (str column, or dict {label: {"from": source}}), or None if unresolvable (e.g. a dict entry with no "from" operator).""" if type(head) is str: return head if type(head) is dict: for _label, operator_line in head.items(): if operator_line and "from" in operator_line: return operator_line["from"] return None def _column_passes_filter(self, head, fields): """True if a raw column head should be kept: columns whose source is in _DATAVIEW_BUILTIN_SOURCES (keyname/classname/is_enabled) and columns with no resolvable source field always pass; other columns (including displayname/last_update/last_sync) are matched against fields via field_export_match().""" source = self._column_source(head) if source is None or source in _DATAVIEW_BUILTIN_SOURCES: return True return field_export_match(source, fields) def get_columns(self, fields=None, action=None): """Return the list of column names from self.content["columns"], for export: the underlying source field name (see _column_source()) rather than the view's display label/rename, so exported files stay importable (a dataview labeling a "keyname"-backed column "Login" still exports a "keyname" header/key, not "Login"). Falls back to the label only when the column has no resolvable source (e.g. a dict entry with no "from"). `fields` (see field_export_match()) drops columns whose underlying instance field doesn't match; builtin columns and columns with no resolvable source field are always kept. `classname`/`keyname` are guaranteed present (prepended if the view's own columns don't already cover them, by source - a renamed `from: keyname` column counts) so the export stays re-importable via `cavaliba load` even if the view itself never displays them. Columns sharing the same source (e.g. two different labels both `from: classname`) are de-duplicated, keeping only the first, to avoid two identical headers. `action`, when given, prepends an "_action" column.""" reply = [] sources_seen = set() raw = self.content.get("columns") or [] for entry in raw: if not self._column_passes_filter(entry, fields): continue source = self._column_source(entry) if source: if source in sources_seen: continue sources_seen.add(source) reply.append(source) elif type(entry) is dict: reply.append(list(entry.keys())[0]) else: reply.append("") if "keyname" not in sources_seen: reply = ["keyname"] + reply if "classname" not in sources_seen: reply = ["classname"] + reply if action: reply = ["_action"] + reply return reply def filter(self, instance=None, mode="datapoint", value_only=False, fields=None, action=None): """ IN: * dataview() object, instance() object * mode: datapoint|value|csv OUT: * array ["val1","val2", {}, {},{} ...] values: * string if keyname/displayname/... * mode=datapoint : datapoint dict {} * mode=value {'fieldname': 'mystring', 'displayname': 'MyString', 'description': 'A string', 'dataformat': 'string', 'dataformat_ext': None, 'is_multi': False, 'bigset': False, 'value': 'data for test1-01' {} for more complex fields like enumerate, user } * mode=csv : value or val1^val2^val3 ... if multi `fields` (see field_export_match()) drops columns whose underlying instance field doesn't match, mirroring get_columns(fields=...) so the returned array stays aligned with its header. For mode in value/csv (export), `classname`/`keyname` are guaranteed present and duplicate-source columns are skipped, mirroring get_columns() - see its docstring - so exported rows stay aligned with get_columns()'s header. mode=datapoint (UI table) skips that guarantee instead, since its header is self.columns, which carries no such guarantee - adding hidden columns there would desync row cells from headers. `action`, when given, prepends its value as the first entry - pass the same `action` used to build get_columns() to stay aligned. """ if not instance: return data = [] sources_seen = set() # get dataview requested columns columns_array = self.content.get("columns") or [] for head in columns_array: if not self._column_passes_filter(head, fields): continue source = self._column_source(head) if source: if source in sources_seen: continue sources_seen.add(source) # no operator, no rename, direct column name if type(head) is str: if head == "keyname": data.append(instance.keyname) elif head == "classname": data.append(instance.classname) elif head == "displayname": data.append(instance.displayname) elif head == "is_enabled": if instance.is_enabled: data.append("X") else: data.append("") elif head == "last_update": data.append(instance.last_update) elif head == "last_sync": data.append(instance.last_sync) else: # existing column if head in instance.fields: cell = "" if mode == "value": cell = instance.fields[head].get_value() elif mode == "datapoint": cell = instance.fields[head].get_datapoint_ui_detail() elif mode == "csv": cell = instance.fields[head].get_csv_cell() data.append(cell) # computed column else: # Strange : non-existent column but no compute primitive ; add anyway (empty) data.append("") # operator, rename, ... # - newcol: <= head is dict (col==head) # from: old_col <= operator_line = operator: operator_value elif type(head) is dict: cell = "" # apply all operators for _nouse, operator_line in head.items(): if operator_line: for operator, operator_value in operator_line.items(): if operator == "from": if operator_value in instance.fields: if mode == "value": cell = instance.fields[operator_value].get_value() elif mode == "datapoint": cell = instance.fields[ operator_value ].get_datapoint_ui_detail() elif mode == "csv": cell = instance.fields[operator_value].get_csv_cell() elif operator_value == "displayname": cell = instance.displayname elif operator_value == "keyname": cell = instance.keyname elif operator_value == "classname": cell = instance.classname elif operator_value == "last_update": cell = instance.last_update elif operator_value == "last_sync": cell = instance.last_sync data.append(cell) else: data.append("") if mode != "datapoint": if "keyname" not in sources_seen: data = [instance.keyname] + data if "classname" not in sources_seen: data = [instance.classname] + data if action: data = [action] + data return data def print(self): """Print a human-readable dump of the dataview (debug helper).""" print("_dataview") print(f" keyname: {self.keyname}") print(f" instance_id: {self.instance_id}") print(f" target_class: {self.target_class}") print(f" is_enabled: {self.is_enabled}") print(f" displayname: {self.displayname}") print(f" columns: {self.columns}") print(f" content: {self.content}")