# (c) cavaliba.com - data - views.py import csv import json import re import yaml #from app_data.data import get_classes from app_data.data import Instance, count_instance from app_data.dataview import dataview_for_request from app_data.eav import eav_from_field # export >> exporter.py # from .exporter import data_yaml_response # from .exporter import data_json_response from app_data.exporter import get_file_content, list_export_folder, send_file from app_data.filestore import uuid_to_queryset from app_data.hook import get_hook from app_data.paginator import paginator_for_request from app_data.permissions import has_schema_read_permission from app_data.revision import revision_add, revision_get from app_data.schema import Schema from app_data.search import get_instance_from_advanced_query, get_instance_from_query, get_query_from_request from app_home.configuration import get_configuration from app_home.log import DEBUG, ERROR, INFO, WARNING, log from app_user.aaa import start_view from django.contrib import messages from django.http import FileResponse, HttpResponse from django.shortcuts import redirect, render from django.utils.translation import gettext as _ # ---------------------------------------------- # list of Classes # ---------------------------------------------- def private(request): context = start_view(request, app="data", view="private", noauth="app_sirene:index", perm="p_data_access", noauthz="app_home:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # page/order for UI # ----------------- # [ [page1, [class1, class2, ...] , [ page2, [...] ] , ... ] paginated = [] pagelist = [] index = {} # page => [class1, class2] default_name = get_configuration("home", "GLOBAL_APPNAME") schemas = Schema.listall() for schema in schemas: # includes is_enabled check if not schema.has_read_permission(aaa=aaa): continue # page comes from dataclass entry ; not dashboard ! page = schema.page if not page: page = default_name if len(page) == 0: page = default_name if page not in index: index[page] = [] pagelist.append(page) schema.count = count_instance(schema.classname) index[page].append(schema) for p in pagelist: paginated.append([p, index[p]]) log(DEBUG, aaa=aaa, app="data", view="private", action="list", status="OK", data=f"{len(schemas)} classes") context["paginated"] = paginated return render(request, 'app_data/private.html', context) # ------------------------------------------------------------------------- # Instance List # ------------------------------------------------------------------------- def instance_list(request, classname=None): context = start_view(request, app="data", view="instance_list", noauth="app_sirene:index", perm="p_data_access", noauthz="app_data:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] schema = Schema.from_name(classname) if not schema: return redirect("app_data:private") max_size = int(get_configuration("data","DATA_DEFAULT_SIZE")) paginator = paginator_for_request(request=request, schema=schema, update_session=True, max_size=max_size) if not paginator: return redirect("app_data:private") query = get_query_from_request(request=request, classname=classname, update_session=True) if not query: query = "" # instances db_instances = get_instance_from_advanced_query(classname=classname, query=query, offset=paginator.offset, limit=paginator.limit) # apply permission & dataview dataview = dataview_for_request(request=request, classname=classname, update_session=True) filtered = [] for iobj in db_instances: instance = Instance(iobj=iobj, expand=True) if not instance.has_read_permission(aaa=aaa): continue iobj.datapoints = dataview.filter(instance=instance, mode="datapoint") iobj.ui_edit = instance.has_update_permission(aaa=aaa) iobj.ui_delete = instance.has_delete_permission(aaa=aaa) iobj.ui_enable_disable = instance.has_update_permission(aaa=aaa) filtered.append(iobj) log(DEBUG, aaa=aaa, app="data", view="instance_list", action="get", status="OK", data=f"{classname}, {len(filtered)} items") # reply context["schema"] = schema context["query"] = query context["dataview"] = dataview context["paginator"] = paginator context["instances"] = filtered context["instances_count"] = len(filtered) # UI buttons/action context["ui_new"] = schema.has_create_permission(aaa=aaa) if "p_data_export" in aaa['perms']: context["ui_export"] = True #context["ui_export"] = schema.has_read_permission(aaa=aaa) if 'p_data_import' in aaa['perms']: if has_schema_read_permission(aaa=aaa): context["ui_texteditor"] = True return render(request, 'app_data/instance_list.html', context) # ------------------------------------------------------------------------- # Instance Detail # ------------------------------------------------------------------------- def instance_detail(request, id=None): ''' ''' context = start_view(request, app="data", view="instance_detail", noauth="app_sirene:index", perm="p_data_access", noauthz="app_data:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] instance = Instance.from_id(id, expand=True) if not instance: messages.add_message(request, messages.ERROR, _("Instance not found")) log(ERROR, aaa=aaa, app="data", view="instance_detail", action="get", status="KO", data=f"Instance not found {id}") return redirect("app_data:private") classname = instance.classname schema = Schema.from_name(classname) # Check PERMISSIONS if not instance.has_read_permission(aaa=aaa): messages.add_message(request, messages.ERROR, _("Not allowed")) log(ERROR, aaa=aaa, app="data", view="instance_detail", action="get", status="KO", data=f"Not allowed on {classname}:{instance.keyname}") return redirect("app_data:private") # Hook to sirene_notify if classname == "sirene_template": hook_notify = request.GET.get("hook", "no") if hook_notify == "notify": return redirect("app_data:instance_edit", instance.id) # Toggle detailled view ? want_detail = request.session.get("ui_want_detail", "no") new_detail = request.GET.get("detail",None) if new_detail: want_detail = new_detail request.session["ui_want_detail"] = want_detail if want_detail == "yes": form_ui = instance.get_dict_for_ui_detail(skip_external=False, skip_injected=False) toggle_detail = "no" else: form_ui = instance.get_dict_for_ui_detail(skip_external=True, skip_injected=False) toggle_detail = "yes" log(DEBUG, aaa=aaa, app="data", view="instance", action="detail", status="OK", data=f"{classname} / {instance.keyname}") context["instance"] = form_ui context["schema"] = schema #context["dataclasses"] = get_classes() # Hook ; inject actions hooks = get_hook(classname=classname, view='detail') context["hooks"] = hooks # UI button/action context["toggle_detail"] = toggle_detail context["ui_edit"] = instance.has_update_permission(aaa=aaa) context["ui_delete"] = instance.has_delete_permission(aaa=aaa) context["ui_security_view"] = "p_data_security_view" in aaa["perms"] context["ui_texteditor"] = 'p_data_import' in aaa['perms'] # revisions - V3.23 context['revisions'] = revision_get(classname=classname, keyname=instance.keyname) if len(context['revisions']) == 0: del context['revisions'] # related objects context['related'] = instance.related() if request.GET.get("print",None): return render(request, 'app_data/instance_print.html', context) elif want_detail == "yes": return render(request, 'app_data/instance_detail_tech.html', context) else: return render(request, 'app_data/instance_detail.html', context) # ------------------------------------------------------------------------- # Instance NEW # ------------------------------------------------------------------------- def instance_new(request, classname=None): ''' Blank form for new instance. POST to save''' context = start_view(request, app="data", view="instance_new", noauth="app_sirene:index", perm="p_data_access", noauthz="app_data:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] schema = Schema.from_name(classname) if not schema: return redirect("app_data:private") # Check PERMISSIONS if not schema.has_create_permission(aaa): messages.add_message(request, messages.ERROR, _("Not allowed")) log(ERROR, aaa=aaa, app="data", view="instance_new", action="post", status="KO", data=f"Not allowed on {classname}") return redirect("app_data:private") else: context["ui_save"] = True # new Data Instance instance = Instance(classname=classname) if request.method == "POST": instance.merge_request(request, aaa=aaa) if instance.is_valid(): r = instance.create() if r: revision_add(aaa=aaa, instance=instance, action="create") messages.add_message(request, messages.SUCCESS, _("Instance created")) log(INFO, aaa=aaa, app="data", view="instance_new", action="post", status="OK", data=f"{classname}/{instance.keyname}") return redirect("app_data:instance_list", classname) else: messages.add_message(request, messages.ERROR, _("Failed to create.")) log(ERROR, aaa=aaa, app="data", view="instance_new", action="post", status="KO", data=f"{classname}") else: err = '; '.join(instance.errors) messages.add_message(request, messages.ERROR, _("Invalid form") + ': ' + err) log(ERROR, aaa=aaa, app="data", view="instance_new", action="post", status="KO", data=f"{err}") form_ui = instance.get_dict_for_ui_form() else: form_ui = instance.get_dict_for_ui_form() log(DEBUG, aaa=aaa, app="data", view="instance_new", action="get", status="OK", data=f"{classname}") context["schema"] = schema context["formular"] = form_ui #context["dataclasses"] = get_classes() context["ui_security_view"] = "p_data_security_view" in aaa["perms"] context["ui_security_edit"] = "p_data_security_edit" in aaa["perms"] return render(request, 'app_data/instance_new.html', context) # ------------------------------------------------------------------------- # Instance EDIT # ------------------------------------------------------------------------- def instance_edit(request, id=None): context = start_view(request, app="data", view="instance_edit", noauth="app_sirene:index", perm="p_data_access", noauthz="app_data:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] instance = Instance.from_id(id) if not instance: messages.add_message(request, messages.ERROR, _("Not found")) log(ERROR, aaa=aaa, app="data", view="instance_detail", action="get", status="KO", data=f"Not found {id}") return redirect("app_data:private") classname = instance.classname schema = Schema.from_name(classname) if not schema: return redirect("app_data:private") # Check PERMISSIONS if not instance.has_update_permission(aaa=aaa): messages.add_message(request, messages.ERROR, _("Not allowed")) log(ERROR, aaa=aaa, app="data", view="instance_edit", action="post", status="KO", data=f"Not allowed on {classname}:{instance.keyname}") return redirect("app_data:private") else: context["ui_save"] = True if request.method == "POST": instance.merge_request(request, aaa=aaa) if instance.is_valid(): r = instance.update() if r: revision_add(aaa=aaa, instance=instance, action="edit") messages.add_message(request, messages.SUCCESS, _("Instance updated")) log(INFO, aaa=aaa, app="data", view="instance_edit", action="post", status="OK", data=f"{classname}:{instance.keyname}") return redirect("app_data:instance_detail", instance.id) else: messages.add_message(request, messages.ERROR, _("Failed to update")) log(ERROR, aaa=aaa, app="data", view="instance_edit", action="post", status="KO", data=f"{classname}:{instance.keyname}") else: err = '; '.join(instance.errors) messages.add_message(request, messages.ERROR, _("Invalid form") + ': ' + err) log(ERROR, aaa=aaa, app="data", view="instance_edit", action="post", status="KO", data=f"{err}") form_ui = instance.get_dict_for_ui_form() else: form_ui = instance.get_dict_for_ui_form() log(DEBUG, aaa=aaa, app="data", view="instance_edit", action="get", status="OK", data=f"{classname} / {instance.keyname}") context["schema"] = schema context["formular"] = form_ui #context["dataclasses"] = get_classes() context["ui_security_view"] = "p_data_security_view" in aaa["perms"] context["ui_security_edit"] = "p_data_security_edit" in aaa["perms"] # context["ui_save"] = True # context["ui_save_as"] = True return render(request, 'app_data/instance_edit.html', context) # ------------------------------------------------------------------------- # Instance DELETE # ------------------------------------------------------------------------- def instance_delete(request, id=None): ''' POST to delete instance by DB id''' context = start_view(request, app="data", view="instance_edit", noauth="app_sirene:index", perm="p_data_access", noauthz="app_data:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] instance = Instance.from_id(id) if not instance: messages.add_message(request, messages.ERROR, _("Instance not found")) log(ERROR, aaa=aaa, app="data", view="instance_delete", action="post", status="KO", data=f"Instance not found {id}") return redirect("app_data:private") classname = instance.classname schema = Schema.from_name(classname) if not schema: return redirect("app_data:private") # Check PERMISSIONS if not instance.has_delete_permission(aaa=aaa): messages.add_message(request, messages.ERROR, _("Not allowed")) log(ERROR, aaa=aaa, app="data", view="instance_delete", action="post", status="KO", data=f"Not allowed on {classname}:{instance.keyname}") return redirect("app_data:private") if request.method == "POST": r = instance.delete() if r: revision_add(aaa=aaa, instance=instance, action="delete") messages.add_message(request, messages.SUCCESS, _("Instance deleted")) log(INFO, aaa=aaa, app="data", view="instance_delete", action="post", status="OK", data=f"{classname}:{instance.keyname}") else: messages.add_message(request, messages.ERROR, _("Failed to delete")) log(ERROR, aaa=aaa, app="data", view="instance_delete", action="post", status="KO", data=f"{classname}:{instance.keyname}") else: messages.add_message(request, messages.ERROR, _("Invalid request")) log(WARNING, aaa=aaa, app="data", view="instance_delete", action="get", status="KO", data="invalid request") context["schema"] = schema return redirect("app_data:instance_list", classname) # ------------------------------------------------------------------------- # Instance ENABLE # ------------------------------------------------------------------------- def instance_enable(request, id=None): context = start_view(request, app="data", view="instance_enable", noauth="app_sirene:index", perm="p_data_access", noauthz="app_data:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] instance = Instance.from_id(id) if not instance: messages.add_message(request, messages.ERROR, _("Instance not found")) log(ERROR, aaa=aaa, app="data", view="instance_enable", action="post", status="KO", data=f"Instance not found {id}") return redirect("app_data:private") classname = instance.classname schema = Schema.from_name(classname) if not schema: return redirect("app_data:private") if not instance.has_update_permission(aaa=aaa): messages.add_message(request, messages.ERROR, _("Not allowed")) log(ERROR, aaa=aaa, app="data", view="instance_enable", action="post", status="KO", data=f"Not allowed on {classname}:{instance.keyname}") return redirect("app_data:private") if request.method == "POST": instance.enable() revision_add(aaa=aaa, instance=instance, action="enable") messages.add_message(request, messages.SUCCESS, _("Instance enabled")) log(INFO, aaa=aaa, app="data", view="instance_enable", action="post", status="OK", data=f"{classname}:{instance.keyname}") else: messages.add_message(request, messages.ERROR, _("Invalid request")) log(WARNING, aaa=aaa, app="data", view="instance_enable", action="get", status="KO", data="invalid request") return redirect("app_data:instance_list", classname) # ------------------------------------------------------------------------- # Instance DISABLE # ------------------------------------------------------------------------- def instance_disable(request, id=None): context = start_view(request, app="data", view="instance_disable", noauth="app_sirene:index", perm="p_data_access", noauthz="app_data:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] instance = Instance.from_id(id) if not instance: messages.add_message(request, messages.ERROR, _("Instance not found")) log(ERROR, aaa=aaa, app="data", view="instance_disable", action="post", status="KO", data=f"Instance not found {id}") return redirect("app_data:private") classname = instance.classname schema = Schema.from_name(classname) if not schema: return redirect("app_data:private") if not instance.has_update_permission(aaa=aaa): messages.add_message(request, messages.ERROR, _("Not allowed")) log(ERROR, aaa=aaa, app="data", view="instance_disable", action="post", status="KO", data=f"Not allowed on {classname}:{instance.keyname}") return redirect("app_data:private") if request.method == "POST": instance.disable() revision_add(aaa=aaa, instance=instance, action="disable") messages.add_message(request, messages.SUCCESS, _("Instance disabled")) log(INFO, aaa=aaa, app="data", view="instance_disable", action="post", status="OK", data=f"{classname}:{instance.keyname}") else: messages.add_message(request, messages.ERROR, _("Invalid request")) log(WARNING, aaa=aaa, app="data", view="instance_disable", action="get", status="KO", data="invalid request") return redirect("app_data:instance_list", classname) # ----------------------------------------- # export tool viewer / producer #----------------------------------------- def data_export(request): context = start_view(request, app="data", view="export", noauth="app_sirene:private", perm="p_data_export", noauthz="app_home:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # GET : export viewer if request.method == "GET": if "p_data_export_viewer" not in aaa['perms']: return redirect("app_home:private") dl_md5 = request.GET.get('dl') view_md5 = request.GET.get('view') context['export_files'] = list_export_folder() # If dl_md5 is provided, download if dl_md5: for export_file in context['export_files']: if export_file['filename_md5'] == dl_md5: response = send_file(export_file['filename']) if response: return response # dl error messages.add_message(request, messages.ERROR, _("Error file not found")) return render(request, 'app_data/export.html', context) # If view_md5 is provided, show elif view_md5: for export_file in context['export_files']: if export_file['filename_md5'] == view_md5: response = get_file_content(export_file['filename']) if response: context["file_content"] = response context["file_name"] = export_file['filename'] return render(request, 'app_data/export.html', context) # view error messages.add_message(request, messages.ERROR, _("Error file not found")) return render(request, 'app_data/export.html', context) else: # display file list return render(request, 'app_data/export.html', context) classname = request.POST.get('classname') m = re.compile(r'^[a-zA-Z0-9()_\/\.\s]*$') if not m.match(classname): return redirect("app_data:private") schema = Schema.from_name(classname) if not schema: return redirect("app_data:private") # Check PERMISSIONS if not schema.has_read_permission(aaa): messages.add_message(request, messages.ERROR, _("Not allowed")) log(ERROR, aaa=aaa, app="data", view="export", action="post", status="KO", data=f"Not allowed on {classname}") return redirect("app_data:private") # max size # NEXT : move to async if too big max_size = int(get_configuration("data","EXPORT_INTERACTIVE_MAX_SIZE")) # page : thispage / allpages page = request.POST.get('page') if page not in ["allpages", "thispage"]: return redirect("app_data:private") if page == "thispage": paginator = paginator_for_request(request=request, schema=schema, max_size=max_size, update_session=False) if not paginator: return redirect("app_data:private") offset = paginator.offset limit = paginator.limit else: offset = 0 limit = max_size # query query = get_query_from_request(request=request, classname=classname, update_session=False) # dataview : name of session dataview or ___ALL___ (no dataview, all cols) if request.POST.get('dv',"") == '___ALL___': dataview = None else: dataview = dataview_for_request(request=request, classname=classname, update_session=True) # csv/yaml/json format = request.POST.get('format') if format not in ["csv","yaml","json"]: return redirect("app_data:private") # refs: comma-separated list of schema names to inline (yaml/json only) raw_refs = request.POST.get('refs', '').strip() refs = [r.strip() for r in raw_refs.split(',') if r.strip()] if raw_refs else None # rev: number of revisions to include (yaml/json only) try: raw_rev = request.POST.get('rev', '').strip() rev = int(raw_rev) if raw_rev else None except (ValueError, TypeError): rev = None db_instances = get_instance_from_advanced_query(classname=classname, query=query, offset=offset, limit=limit) # TODO: check permission on each instance # db_instances = [] # raw_db_instances = get_instance_from_query(classname=classname, query=query, offset=offset, limit=limit) # for i in raw_db_instances: # if has_read_permission_on_instance(aaa=aaa, iobj=i): # db_instances.append(i) count = len(db_instances) # warn if max size reached if count == max_size: log(WARNING, aaa=aaa, app="data", view="export", action="post", status="KO", data=f"Export may be truncated for {classname} - {max_size}") messages.add_message(request, messages.WARNING, _("Export may be truncated (size max reached)")) if dataview: columns_array = dataview.get_columns() if format == "yaml": datalist = [] for iobj in db_instances: instance = Instance(iobj=iobj, expand=True) if not instance: continue if not instance.has_read_permission(aaa=aaa): continue if dataview: datapoints = dataview.filter(instance=instance, mode='value') data = dict(zip(columns_array, datapoints)) else: data = instance.get_dict_for_export(refs=refs, rev=rev) datalist.append(data) filedata = yaml.dump(datalist, allow_unicode=True, Dumper=MyYamlDumper, sort_keys=False) response = HttpResponse(filedata, content_type='text/yaml') response['Content-Disposition'] = 'attachment; filename="data.yaml"' return response if format == "json": datalist = [] for iobj in db_instances: instance = Instance(iobj=iobj, expand=True) if not instance: continue if not instance.has_read_permission(aaa=aaa): continue if dataview: datapoints = dataview.filter(instance=instance, mode='value') data = dict(zip(columns_array, datapoints)) else: data = instance.get_dict_for_export(refs=refs, rev=rev) datalist.append(data) filedata = json.dumps(datalist, indent=4, ensure_ascii=False) response = HttpResponse(filedata, content_type='text/json') response['Content-Disposition'] = 'attachment; filename="data.json"' return response if format == "csv": delimiter = get_configuration(keyname="CSV_DELIMITER") response = HttpResponse(content_type='text/csv') response['Content-Disposition'] = 'attachment; filename="data.csv"' writer = csv.writer(response, delimiter=delimiter) # try: # instance = Instance(iobj=db_instances[0], expand=True) # except Exception: # return response csv_columns = None if dataview: csv_columns = dataview.get_columns() writer.writerow(csv_columns) for iobj in db_instances: instance = Instance(iobj=iobj, expand=True) if not instance: continue if not csv_columns: csv_columns = instance.get_csv_columns() writer.writerow(csv_columns) if not instance.has_read_permission(aaa=aaa): continue if dataview: line = dataview.filter(instance=instance, mode='csv') else: line = instance.get_csv_line(csv_columns) writer.writerow(line) return response context["ui_export"] = True return render(request, 'app_data/export.html', context) #YAML 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() # ------------------------------------------------------------------------- # file_display # ------------------------------------------------------------------------- # v3.19 def file_display(request, fileid=None): ''' Blank form for new instance. POST to save''' context = start_view(request, app="data", view="file_display", noauth="app_sirene:index", perm="p_data_access", noauthz="app_data:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # check perm from EAV parent object eav = eav_from_field(format='file', value=fileid) if not eav: return redirect("app_data:private") classname = eav.classname keyname = eav.keyname instance = Instance.from_keyname(classname=classname, keyname=keyname) if not instance: return redirect("app_data:private") if not instance.has_read_permission(aaa=aaa): return redirect("app_data:private") # get filename and filepath qs = uuid_to_queryset(fileid) if not qs: return redirect("app_data:private") filename = qs.filename filepath = qs.filepath # # get filepath # filepath = uuid_to_filestore_path(fileid) if not filepath: return redirect("app_data:private") return FileResponse(open(filepath, "rb"), filename=filename)