# (c) cavaliba.com - data - views.py """ Core data-browsing views: schema list, instance list/detail/CRUD (new/edit/ delete/enable/disable), 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. Export lives in app_data.views_exporter (the Exporter tool and the export-folder Download page), with its logic in app_data.exporter. 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 File download: file_display """ from django.conf import settings from django.contrib import messages from django.http import FileResponse 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_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.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, query_from_dataview, set_query_in_session, ) 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")) # get dataview - resolved before the query, since a dataview can carry its # own query (loaded into the search box when the user switches to that view) # and that has to be settled before the page reset and the instance fetch # below. dataview = dataview_for_request(request=request, classname=classname, update_session=True) dv_query = query_from_dataview(dataview=dataview, request=request) # 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) links never carry "query", so they're unaffected - but a # dataview-switch (?dv=name) link does become a new filter when that view # has its own query, hence dv_query here. # # 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 or dv_query: 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 = "" # the dataview the user just switched to brings its own query - it replaces # whatever was in the box, and is remembered like a typed one so it survives # the next plain GET (pagination, a detail round-trip) and feeds the # Exporter link if dv_query: query = dv_query set_query_in_session(request, classname, query) # instances db_instances = get_instance_from_advanced_query( classname=classname, query=query, offset=paginator.offset, limit=paginator.limit ) 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) # 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, 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") form_ui = instance.get_dict_for_ui_detail(skip_external=True, skip_injected=False) 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["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) 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/_ipv4.html link in list and detail display_mode, 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) # ------------------------------------------------------------------------- # 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)