# (c) cavaliba.com - data - views.py import csv import json import re import yaml from django.contrib import messages from django.http import FileResponse, HttpResponse from django.shortcuts import redirect, render from django.utils.translation import gettext as _ from app_data.aaa import start_view # from app_data.data import get_classes from app_data.data import 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 inject_iam_group_detail, inject_iam_role_detail, inject_iam_user_detail 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_query_from_request from app_home.configuration import get_configuration from app_home.log import DEBUG, ERROR, INFO, WARNING, log # ---------------------------------------------- # 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 = Schema.count_instances(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", ) # USER hook if classname == "user": context["hook_user"] = True context["ui_user_email_test"] = "p_user_email_test" in aaa["perms"] context["ui_user_sms_test"] = "p_user_sms_test" in aaa["perms"] context["ui_iam_impersonate"] = "p_iam_impersonate" in aaa["perms"] # SIRENE NOTIFY hook : direct access to notifications from action menu if schema.notify and "p_sirene_new" in aaa.get("perms", []): context["hook_schema_notify"] = True context["ui_sirene_new"] = "p_sirene_new" in aaa.get("perms", []) # reply context["schema"] = schema context["query"] = query context["dataview"] = dataview context["paginator"] = paginator context["instances"] = filtered context["instances_count"] = len(filtered) # UI perms context["ui_new"] = schema.has_create_permission(aaa=aaa) context["ui_export"] = "p_data_export" in aaa["perms"] 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") # 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 # HOOKS: inject actions / content # USER hook if classname == "user": context["hook_user"] = True context["ui_user_email_test"] = "p_user_email_test" in aaa["perms"] context["ui_user_sms_test"] = "p_user_sms_test" in aaa["perms"] context["ui_iam_impersonate"] = "p_iam_impersonate" in aaa["perms"] inject_iam_user_detail(instance, context) # GROUP hook elif classname == "group": context["hook_group"] = True context["hook_group"] = True inject_iam_group_detail(instance, context) # ROLE hook elif classname == "role": context["hook_role"] = True inject_iam_role_detail(instance, context) # SIRENE NOTIFY hook : direct access to notifications from action menu if schema.notify and "p_sirene_new" in aaa.get("perms", []): context["hook_schema_notify"] = True context["ui_sirene_new"] = "p_sirene_new" in aaa.get("perms", []) # 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() # TODO: merge templates 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["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["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)