# (c) cavaliba.com - data - views.py """ Core data-browsing views: schema list, instance list/detail/CRUD (new/edit/ delete/enable/disable), export (viewer + CSV/YAML/JSON producer), and file download. All views go through app_data.aaa.start_view for auth/permission gating; permission checks are then re-verified per-schema/per-instance. Method inventory: Schema/instance browsing: private, instance_list, instance_detail, ipam_ip_jump Instance CRUD: instance_new, instance_duplicate, instance_edit, instance_delete, instance_enable, instance_disable Export: data_export, MyYamlDumper.write_line_break File download: file_display """ import csv import io import json import re import yaml from django.conf import settings from django.contrib import messages from django.http import FileResponse, HttpResponse from django.shortcuts import redirect, render from django.urls import reverse 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, 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.fieldtypes.field_ipv4 import check_valid_ipv4 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, inject_ipam_ip_detail, inject_ipam_subnet_detail, inject_ipam_vlan_detail, ) from app_data.ipam import build_subnet_rows, build_vlan_rows, hexip from app_data.paginator import paginator_for_request from app_data.permissions import has_schema_read_permission from app_data.revision import 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_schema_order(schemas): """Reorder schemas per settings.SCHEMA_LIST_ORDER: "^name" pins a schema first, "$name" pins a schema last (both in the order listed), "-name" drops a schema from the list entirely. Every other schema is inserted alphabetically (by classname/keyname) between the pinned-first and pinned-last groups.""" first_names = [] last_names = [] hidden_names = set() for entry in settings.SCHEMA_LIST_ORDER: if not entry or entry[0] not in "^$-": continue prefix, name = entry[0], entry[1:] if prefix == "^": first_names.append(name) elif prefix == "$": last_names.append(name) else: hidden_names.add(name) by_classname = {schema.classname: schema for schema in schemas} first = [ by_classname[name] for name in first_names if name in by_classname and name not in hidden_names ] last = [ by_classname[name] for name in last_names if name in by_classname and name not in hidden_names ] pinned_names = set(first_names) | set(last_names) middle = [ schema for schema in schemas if schema.classname not in pinned_names and schema.classname not in hidden_names ] middle.sort(key=lambda schema: schema.classname.lower()) return first + middle + last def private(request): """List all schemas the user can read, ordered per settings.SCHEMA_LIST_ORDER (pin first/last, hide) with the rest sorted alphabetically. Instance counts are computed live (not cached).""" context = start_view( request, app="data", view="private", perm="p_data_access", noauthz="app_home:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] visible = [] for schema in Schema.listall(): # includes is_enabled check if not schema.has_read_permission(aaa=aaa): continue schema.count = Schema.count_instances(schema.classname) visible.append(schema) visible = _private_schema_order(visible) log(DEBUG, aaa=aaa, app="data", view="private", action="list", status="OK", data=f"{len(visible)} classes") # fmt: skip context["schemas"] = visible return render(request, "app_data/private.html", context) # ------------------------------------------------------------------------- # Instance List # ------------------------------------------------------------------------- def instance_list(request, classname=None): """Paginated, searchable, dataview-filtered list of instances for a schema, with per-instance read/update/delete permission flags for the UI.""" context = start_view( request, app="data", view="instance_list", perm="p_data_access", noauthz="app_home:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] schema = Schema.from_name(classname) if not schema: return redirect("app_home:private") # max_size = int(get_configuration("data", "DATA_DEFAULT_SIZE")) # the search box (cv-search-form) POSTs a new query to this same view, and # some links elsewhere (instance_detail related-objects badges, ipam_subnet # CIDR links) jump in with an explicit "?query=..." GET param - either way # it's a new filter, so start back at page 1: paginator_for_count() only # clamps against the schema's total instance count, not the query-matched # count, so a narrowing filter can otherwise leave a remembered page 2/3/... # showing nothing (offset past the filtered result set). Plain pagination # (?page=N&size=M) and dataview-switch (?dv=name) links never carry # "query", so they're unaffected. # # cv-search-form has action="" - submitting it from ".../list/?page=3..." # POSTs back to that very URL (empty relative reference keeps the current # query string, RFC 3986 5.3), so request.GET still carries the stale # "page=3" even though this is a POST. paginator_for_count() reads "page" # from request.GET before the session, so writing the session alone # doesn't override it - the stale GET param must be dropped too. if request.method == "POST" or "query" in request.GET: request.session[classname + "_page"] = 1 if "page" in request.GET: request.GET = request.GET.copy() del request.GET["page"] paginator = paginator_for_request(request=request, schema=schema, update_session=True) if not paginator: return redirect("app_home: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 ) # get 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") # fmt: skip # IPAM SUBNET hook: hardcoded columns (keyname, description, size, occupancy, # VLAN - same as the instance_detail Subnet Hierarchy table) instead of the # dataview-driven ones, but only when the default/builtin dataview is active - # a user-picked custom dataview is respected as normal. A hack for now: no # generic way for a dataview to request computed/external columns yet. if classname == "ipam_subnet" and dataview.keyname == "ipam_subnet_default": context["hook_ipam_subnet_list"] = True subnets = [] for iobj in filtered: subnets.append( {"id": iobj.id, "keyname": iobj.keyname, "displayname": iobj.displayname} ) rows = build_subnet_rows(subnets, aaa=aaa, with_subnet_detail=True) rows_by_id = {} for row in rows: rows_by_id[row["id"]] = row # build_subnet_rows() silently drops instances with a malformed # (unaligned) CIDR keyname - drop them here too, not just their # computed columns, otherwise _subnet_hierarchy_row.html renders # with row=None and {% url ... row.id %} blows up on the empty id. filtered = [iobj for iobj in filtered if iobj.id in rows_by_id] for iobj in filtered: iobj.ipam_subnet_row = rows_by_id[iobj.id] # IPAM VLAN hook: hardcoded columns (keyname, description, VLAN ID, # subnet(s) with size/occupancy - same computed data as the # instance_detail VLAN hook, minus the ancestor climbing) instead of the # dataview-driven ones, but only when the default/builtin dataview is # active - same hack as the ipam_subnet hook above. if classname == "ipam_vlan" and dataview.keyname == "ipam_vlan_default": context["hook_ipam_vlan_list"] = True vlans = [ {"id": iobj.id, "keyname": iobj.keyname, "displayname": iobj.displayname} for iobj in filtered ] rows = build_vlan_rows(vlans, aaa=aaa) rows_by_id = {row["id"]: row for row in rows} for iobj in filtered: iobj.ipam_vlan_row = rows_by_id[iobj.id] # 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) context["supported_encoding"] = settings.SUPPORTED_ENCODING context["default_csv_delimiter"] = get_configuration(keyname="CSV_DELIMITER") # UI perms context["ui_new"] = schema.has_create_permission(aaa=aaa) context["ui_export"] = "p_data_export" in aaa["perms"] if "p_code_editor" in aaa["perms"]: if has_schema_read_permission(aaa=aaa): context["ui_code_editor"] = True return render(request, "app_data/instance_list.html", context) # ------------------------------------------------------------------------- # Instance Detail # ------------------------------------------------------------------------- def instance_detail(request, id=None): """Show a single instance's fields (toggleable technical/detail view), plus user/group/role hooks, revisions, and related-object list.""" context = start_view( request, app="data", view="instance_detail", perm="p_data_access", noauthz="app_home: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}") # fmt: skip return redirect("app_home: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}") # fmt: skip return redirect("app_home: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}") # fmt: skip 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) # IPAM IP hook elif classname == "ipam_ip": context["hook_ipam_ip"] = True inject_ipam_ip_detail(instance, context, aaa=aaa) # IPAM SUBNET hook elif classname == "ipam_subnet": context["hook_ipam_subnet"] = True inject_ipam_subnet_detail(instance, context, aaa=aaa) # IPAM VLAN hook elif classname == "ipam_vlan": context["hook_ipam_vlan"] = True inject_ipam_vlan_detail(instance, context, aaa=aaa) # 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_duplicate"] = schema.has_create_permission(aaa=aaa) context["ui_security_view"] = "p_data_security_view" in aaa["perms"] context["ui_code_editor"] = "p_code_editor" 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(aaa=aaa) # 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) # ------------------------------------------------------------------------- # IPAM IP jump - ?ip=X -> that IP's ipam_ip instance_detail page # ------------------------------------------------------------------------- def ipam_ip_jump(request): """Redirect an ipv4-format field value (fields/_datapoint_detail.html / _datapoint_list.html link, used for any ipv4 field on any schema) to that IP's ipam_ip instance_detail page. Read-only: does not create anything - if no ipam_ip entry exists yet for this IP (e.g. it predates the real-time auto-discovery hook, or housekeeping's ipam_refresh() hasn't run since), falls back to the ipam_ip list prefiltered by the IP instead. ipam_ip now has a full instance_detail page (related objects, subnet/VLAN hierarchy via its hook) so this is the single generic destination - replaces app_ipam's old ip_view.""" context = start_view( request, app="data", view="ipam_ip_jump", perm="p_data_access", ) if context["redirect"]: return redirect(context["redirect"]) ip = request.GET.get("ip", "").strip() if not check_valid_ipv4(ip): messages.add_message(request, messages.ERROR, _("Not a valid IPv4 address.")) return redirect("app_data:instance_list", classname="ipam_ip") hexkey = hexip(ip) instance = Instance.from_keyname(classname="ipam_ip", keyname=hexkey) if instance: return redirect("app_data:instance_detail", id=instance.id) return redirect( f"{reverse('app_data:instance_list', kwargs={'classname': 'ipam_ip'})}?query={ip}" ) # ------------------------------------------------------------------------- # 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", perm="p_data_access", noauthz="app_home:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] schema = Schema.from_name(classname) if not schema: return redirect("app_home: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}") # fmt: skip return redirect("app_home: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.save(aaa=aaa, action="create") if r: 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}") # fmt: skip 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}") # fmt: skip 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}") # fmt: skip form_ui = instance.get_dict_for_ui_form() else: # optional prefill of the new instance's keyname from the calling page # (e.g. a "Create" link offered from a lookup that already knows the value) prefill_keyname = request.GET.get("keyname") if prefill_keyname: instance.keyname = prefill_keyname form_ui = instance.get_dict_for_ui_form() log(DEBUG, aaa=aaa, app="data", view="instance_new", action="get", status="OK", data=f"{classname}") # fmt: skip 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) def instance_duplicate(request, id=None): """GET only: build a blank new-instance form prefilled from an existing instance's field values (keyname excluded). Schema is derived from the source instance itself; actual creation POSTs through instance_new.""" context = start_view( request, app="data", view="instance_duplicate", perm="p_data_access", noauthz="app_home:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] source = Instance.from_id(id) if not source: messages.add_message(request, messages.ERROR, _("Instance not found")) log(ERROR, aaa=aaa, app="data", view="instance_duplicate", action="get", status="KO", data=f"Instance not found {id}") # fmt: skip return redirect("app_home:private") classname = source.classname schema = Schema.from_name(classname) if not schema: return redirect("app_home:private") if not source.has_read_permission(aaa=aaa): messages.add_message(request, messages.ERROR, _("Not allowed")) log(ERROR, aaa=aaa, app="data", view="instance_duplicate", action="get", status="KO", data=f"Not allowed to read {classname}:{source.keyname}") # fmt: skip return redirect("app_home:private") if not schema.has_create_permission(aaa): messages.add_message(request, messages.ERROR, _("Not allowed")) log(ERROR, aaa=aaa, app="data", view="instance_duplicate", action="get", status="KO", data=f"Not allowed to create {classname}") # fmt: skip return redirect("app_home:private") new_instance = Instance(classname=classname) data = source.to_dict() data.pop("keyname", None) new_instance.merge_import(data, aaa=aaa) context["ui_save"] = True context["schema"] = schema context["formular"] = new_instance.get_dict_for_ui_form() context["ui_security_view"] = "p_data_security_view" in aaa["perms"] context["ui_security_edit"] = "p_data_security_edit" in aaa["perms"] log(INFO, aaa=aaa, app="data", view="instance_duplicate", action="get", status="OK", data=f"{classname}/{source.keyname}") # fmt: skip return render(request, "app_data/instance_new.html", context) # ------------------------------------------------------------------------- # Instance EDIT # ------------------------------------------------------------------------- def instance_edit(request, id=None): """Edit form for an existing instance. GET shows the form; POST validates and saves via instance.save().""" context = start_view( request, app="data", view="instance_edit", perm="p_data_access", noauthz="app_home: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}") # fmt: skip return redirect("app_home:private") classname = instance.classname schema = Schema.from_name(classname) if not schema: return redirect("app_home: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}") # fmt: skip return redirect("app_home:private") else: context["ui_save"] = True if request.method == "POST": instance.merge_request(request, aaa=aaa) if instance.is_valid(): r = instance.save(aaa=aaa, action="update") if r: 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}") # fmt: skip 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}") # fmt: skip 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}") # fmt: skip 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}") # fmt: skip 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_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_delete", perm="p_data_access", noauthz="app_home: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}") # fmt: skip return redirect("app_home:private") classname = instance.classname schema = Schema.from_name(classname) if not schema: return redirect("app_home: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}") # fmt: skip return redirect("app_home:private") if request.method == "POST": r = instance.delete(aaa=aaa) if r: 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}") # fmt: skip 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}") # fmt: skip 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") # fmt: skip context["schema"] = schema return redirect("app_data:instance_list", classname) # ------------------------------------------------------------------------- # Instance ENABLE # ------------------------------------------------------------------------- def instance_enable(request, id=None): """POST to enable an instance by DB id.""" context = start_view( request, app="data", view="instance_enable", perm="p_data_access", noauthz="app_home: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}") # fmt: skip return redirect("app_home:private") classname = instance.classname schema = Schema.from_name(classname) if not schema: return redirect("app_home: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}") # fmt: skip return redirect("app_home:private") if request.method == "POST": instance.enable(aaa=aaa) 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}") # fmt: skip 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") # fmt: skip return redirect("app_data:instance_list", classname) # ------------------------------------------------------------------------- # Instance DISABLE # ------------------------------------------------------------------------- def instance_disable(request, id=None): """POST to disable an instance by DB id.""" context = start_view( request, app="data", view="instance_disable", perm="p_data_access", noauthz="app_home: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}") # fmt: skip return redirect("app_home:private") classname = instance.classname schema = Schema.from_name(classname) if not schema: return redirect("app_home: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}") # fmt: skip return redirect("app_home:private") if request.method == "POST": instance.disable(aaa=aaa) 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}") # fmt: skip 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") # fmt: skip return redirect("app_data:instance_list", classname) # ----------------------------------------- # export tool viewer / producer # ----------------------------------------- def data_export(request): """GET: export viewer (list/download/preview previously exported files). POST: produce a csv/yaml/json export of a schema's instances (this page or all pages, dataview-filtered or all columns, with optional refs/rev).""" context = start_view( request, app="data", view="export", 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_home:private") schema = Schema.from_name(classname) if not schema: return redirect("app_home: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}") # fmt: skip return redirect("app_home:private") # max size # NEXT : move to async if too big max_size = int(get_configuration("data", "EXPORT_INTERACTIVE_MAX_SIZE")) # always export up to max_size (no "current page only" mode - the export # modal never sends anything but "allpages") 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_home:private") # refs: comma-separated list of schema names to inline (yaml/json only). # Only reference fields actually selected for export contribute to _refs: # all fields passing `fields` under "Columns: All", or only the active # dataview's own columns under "Current View" - see # Instance.get_export_refs()'s `selected_fields` param. 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 # fields: filter which instance fields to export - "a,b,c*" / "*" / "!a,b,c*" # (see app_data.data.field_export_match()) - applies both to "All columns" # and to a named dataview's columns (matched against each column's source field) fields = request.POST.get("fields", "").strip() or None db_instances = get_instance_from_advanced_query( classname=classname, query=query, offset=offset, limit=limit ) 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}") # fmt: skip messages.add_message( request, messages.WARNING, _("Export may be truncated (size max reached)") ) # dv_fields: free-form field-list override for a dataview export, applied # uniformly across CSV/YAML/JSON - when set, it replaces the active # dataview's own declared columns with exactly the typed field names # (need not already be part of the view); left blank, all three formats # fall back to the view's own columns, as declared. dv_fields_text = request.POST.get("dv_fields", "").strip() or None dv_override_columns = DataView.parse_dv_fields(dv_fields_text) if dataview else None if dataview: columns_array = dataview.get_columns(fields=fields, override_columns=dv_override_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", fields=fields, override_columns=dv_override_columns, ) data = dict(zip(columns_array, datapoints, strict=True)) if refs: refs_dict = instance.get_export_refs(refs=refs, selected_fields=columns_array) if refs_dict: data["_refs"] = refs_dict if rev: revlist = instance.get_export_revisions(rev=rev) if revlist: data["_revision"] = revlist else: data = instance.get_dict_for_export(refs=refs, rev=rev, fields=fields) 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", fields=fields, override_columns=dv_override_columns, ) data = dict(zip(columns_array, datapoints, strict=True)) if refs: refs_dict = instance.get_export_refs(refs=refs, selected_fields=columns_array) if refs_dict: data["_refs"] = refs_dict if rev: revlist = instance.get_export_revisions(rev=rev) if revlist: data["_revision"] = revlist else: data = instance.get_dict_for_export(refs=refs, rev=rev, fields=fields) 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 = request.POST.get("delimiter", "").strip() or get_configuration( keyname="CSV_DELIMITER" ) if len(delimiter) != 1: messages.add_message( request, messages.ERROR, _("CSV delimiter must be a single character") ) return redirect("app_home:private") encoding = request.POST.get("encoding", "").strip() or "utf-8" buffer = io.StringIO() writer = csv.writer(buffer, delimiter=delimiter) # try: # instance = Instance(iobj=db_instances[0], expand=True) # except Exception: # return response # dataview CSV export is otherwise a pure passthrough: real field names # from the view (or the dv_fields textbox override, same # dv_override_columns/dv_fields_text as the YAML/JSON branches above), # deduped, no forced classname/id/keyname and no fields-filter grammar - # see DataView.get_csv_export_columns()/get_csv_export_row(). Both CSV # paths (dataview and "Columns: All", via Instance.get_csv_columns()/ # get_csv_line()) always start with "_schema" though, so the export # stays classname-tagged/round-trippable through cavaliba load/the # Import Tool. csv_columns = None if dataview: csv_columns = dataview.get_csv_export_columns(dv_fields_text) writer.writerow(csv_columns) for iobj in db_instances: instance = Instance(iobj=iobj, expand=True) if not instance: continue if csv_columns is None: csv_columns = instance.get_csv_columns(fields=fields) writer.writerow(csv_columns) if not instance.has_read_permission(aaa=aaa): continue if dataview: line = dataview.get_csv_export_row(instance, csv_columns) else: line = instance.get_csv_line(csv_columns) writer.writerow(line) try: raw = buffer.getvalue().encode(encoding) except LookupError: messages.add_message(request, messages.ERROR, _("Unknown CSV encoding: ") + encoding) return redirect("app_home:private") except UnicodeEncodeError: messages.add_message( request, messages.ERROR, _("Some data cannot be encoded as ") + encoding ) return redirect("app_home:private") response = HttpResponse(raw, content_type=f"text/csv; charset={encoding}") response["Content-Disposition"] = 'attachment; filename="data.csv"' 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): """Emit an extra blank line after top-level YAML blocks for readability.""" super().write_line_break(data) if len(self.indents) < 2: super().write_line_break() # ------------------------------------------------------------------------- # file_display # ------------------------------------------------------------------------- # v3.19 def file_display(request, fileid=None): """Stream a stored file by its UUID, gated on read permission of the instance that owns it (resolved via its DataEAV "file" entry).""" context = start_view( request, app="data", view="file_display", perm="p_data_access", noauthz="app_home: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_home:private") classname = eav.classname keyname = eav.keyname instance = Instance.from_keyname(classname=classname, keyname=keyname) if not instance: return redirect("app_home:private") if not instance.has_read_permission(aaa=aaa): return redirect("app_home:private") # get filename and filepath qs = uuid_to_queryset(fileid) if not qs: return redirect("app_home:private") filename = qs.filename filepath = qs.filepath # # get filepath # filepath = uuid_to_filestore_path(fileid) if not filepath: return redirect("app_home:private") return FileResponse(open(filepath, "rb"), filename=filename)