# (c) cavaliba.com - data - views_importer.py """ The Import Tool page: a single form for CSV/YAML/JSON file import. Optionally runs entries through a Pipeline, then either just verifies parsing or imports via load_broker (run_import_batch()). Files whose row count exceeds settings.CAVALIBA_MAX_SYNC_IMPORT are handed off to an async DataTask (app_data.tasks.submit_import) instead of being written inside the request. Method inventory: import_tool. Private helpers: _run_and_report, _report_import, _page_size_to_first_last. """ from django.conf import settings from django.contrib import messages from django.shortcuts import redirect, render from django.utils.translation import gettext as _ from app_data.aaa import start_view from app_data.filestore import tmp_local_filepath from app_data.forms import ImportForm from app_data.loader import load_file_csv, load_file_json, load_file_yaml, run_import_batch from app_data.pipeline import Pipeline from app_data.schema import Schema from app_data.tasks import submit_import from app_home.log import DEBUG, ERROR, INFO, WARNING, log _LOADER_BY_KIND = {"csv": load_file_csv, "yaml": load_file_yaml, "json": load_file_json} # ------------------------------------------------------------------------- # IMPORT TOOL (single form: CSV/YAML/JSON) # ------------------------------------------------------------------------- def import_tool(request): """GET: render the Import Tool page. POST: parse+run, then either re-render in place (verify / parse error / invalid form) or redirect (import: home for a sync run, the task detail page for a large async run).""" context = start_view( request, app="data", view="import_tool", noauth="app_sirene:private", perm="p_data_import", noauthz="app_home:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # NOTE: is_data_admin (the "p_data_admin" permission) gates the Advanced # section (force_action/force_schema) - it is a different, narrower # concept than aaa["is_admin"] (the broader admin/role_admin flag). Do # not conflate the two here. is_data_admin = "p_data_admin" in aaa["perms"] pipelines = Pipeline.list(is_enabled=True) schemas = [s for s in Schema.listall() if s.has_create_permission(aaa)] if request.method == "POST": form = ImportForm( request.POST, request.FILES, pipelines=pipelines, schemas=schemas, is_admin=is_data_admin, ) if form.is_valid(): response = _run_and_report(request, aaa, form, is_data_admin) if response: return response else: messages.add_message(request, messages.ERROR, _("Import failed - invalid form")) else: form = ImportForm(pipelines=pipelines, schemas=schemas, is_admin=is_data_admin) context["form"] = form context["is_data_admin"] = is_data_admin return render(request, "app_data/import.html", context) def _run_and_report(request, aaa, form, is_data_admin): """Save the upload, parse it by extension (.csv/.yml/.yaml/.json), then either verify (row count only, no DB writes) or import. Large imports (row count over settings.CAVALIBA_MAX_SYNC_IMPORT) are handed off to an async DataTask instead of running the write loop inside this request - refused upfront if the requester lacks p_task_view, since they'd have no way to see that task's progress. Returns None to stay on the form (re-rendered by the caller with the bound - and so still filled in - form), or a ready-made redirect response.""" cleaned = form.cleaned_data postfile = cleaned["file"] pipeline_name = cleaned.get("pipeline") or None pipeline = Pipeline.from_name(pipeline_name) if pipeline_name else None # force_action/force_schema/sync_mode are only ever present in # cleaned_data for an admin form instance (ImportForm.__init__ deletes # all three fields otherwise) - the explicit is_data_admin check here is # defense in depth, keeping the authorization rule visible at the call # site. force_action = cleaned.get("force_action") or None if is_data_admin else None force_schema = cleaned.get("force_schema") or None if is_data_admin else None # aaa["sync_mode"] defaults to False (see aaa.py) - Instance.save() reads # it to decide whether to touch last_sync, same knob as ?sync=true on # /api/load/ and /api/rawfile/. Mutating aaa in place here means it's # already set on the same dict submit_import() forwards to the async task. aaa["sync_mode"] = bool(cleaned.get("sync_mode")) if is_data_admin else False first, last = _page_size_to_first_last(cleaned.get("page"), cleaned.get("size")) filename = tmp_local_filepath() with open(filename, "wb+") as destination: for chunk in postfile.chunks(): destination.write(chunk) name = postfile.name.lower() if name.endswith(".csv"): file_kind = "csv" parse_options = { "encoding": cleaned.get("encoding") or "utf-8", "csv_delimiter": cleaned.get("separator") or ",", "no_multi": not cleaned.get("split_multivalue"), "first": first, "last": last, } elif name.endswith(".yml") or name.endswith(".yaml"): file_kind = "yaml" parse_options = {"first": first, "last": last} elif name.endswith(".json"): file_kind = "json" parse_options = {"first": first, "last": last} else: messages.add_message(request, messages.ERROR, _("Import failed - unsupported file type")) return None datalist, err = _LOADER_BY_KIND[file_kind](filename=filename, **parse_options) if err: messages.add_message(request, messages.ERROR, _("Import failed") + f" - {err}") log(ERROR, aaa=aaa, app="data", view="import", action="file", status="KO", data=f"{err} : {postfile.name}") # fmt: skip return None submit = request.POST.get("submit") large_import = len(datalist) > settings.CAVALIBA_MAX_SYNC_IMPORT if submit == "verify": note = "" if large_import: note = " - " + _("Import will run as a background task for this file") messages.add_message(request, messages.SUCCESS, _("Check ok") + f" ({len(datalist)} rows)" + note) # fmt: skip log(DEBUG, aaa=aaa, app="data", view="import", action="check", status="OK") return None if submit == "import": if large_import: if "p_task_view" not in aaa["perms"]: messages.add_message( request, messages.ERROR, _("Import failed - file too large for a live import") + f" ({len(datalist)} rows, max {settings.CAVALIBA_MAX_SYNC_IMPORT}) - " + _( "ask an administrator for the p_task_view permission to run large imports in the background" ), # noqa: E501 ) log(WARNING, aaa=aaa, app="data", view="import", action="import", status="DENY", data=f"large import refused, missing p_task_view ({len(datalist)} rows)") # fmt: skip return None handle = submit_import( filename=filename, file_kind=file_kind, parse_options=parse_options, aaa=aaa, pipeline_name=pipeline_name, force_action=force_action, force_schema=force_schema, owner_type="user", owner_id=aaa["username"], ) messages.add_message( request, messages.SUCCESS, _("Large import ({n} rows) started as a background task").format(n=len(datalist)), ) log(INFO, aaa=aaa, app="data", view="import", action="import", status="OK", data=f"async import submitted, handle={handle}, rows={len(datalist)}") # fmt: skip return redirect("app_data:task_detail", handle=handle) result = run_import_batch( datalist, aaa, pipeline=pipeline, force_action=force_action, force_schema=force_schema, max_failure=settings.CAVALIBA_MAX_IMPORT_ERROR, ) _report_import(request, aaa, result) return redirect("app_home:private") return None def _report_import(request, aaa, result): """Report import results: ok/ko counts, an early-abort note if the failure count crossed CAVALIBA_MAX_IMPORT_ERROR, and the first 20 error messages. Shared by CSV/YAML/JSON alike (result comes from loader.run_import_batch()).""" count_ok = result["count_ok"] count_ko = result["count_ko"] errors = result["errors"] aborted = result["aborted"] if aborted: messages.add_message( request, messages.ERROR, _("Import aborted - too many failures") + f" (ok={count_ok}, ko={count_ko}, max={settings.CAVALIBA_MAX_IMPORT_ERROR})", ) elif count_ko: messages.add_message( request, messages.ERROR, _("Import partial or failed") + f" (ok={count_ok}, ko={count_ko})", ) else: messages.add_message(request, messages.SUCCESS, _("Import OK") + f" ({count_ok})") for err in errors[:20]: messages.add_message(request, messages.ERROR, str(err)) log(INFO, aaa=aaa, app="data", view="import", action="import", status="KO" if (count_ko or aborted) else "OK", data=f"ok={count_ok} ko={count_ko} aborted={aborted}") # fmt: skip def _page_size_to_first_last(page, size): """Translate optional page/size (already validated by ImportForm.clean() - page without size is rejected there, so it never reaches here) into loader-style 1-based inclusive first/last. size alone implies page=1 (first N rows); both blank means no limit (1, 0), matching the loader's own defaults.""" if size: page = page or 1 return (page - 1) * size + 1, page * size return 1, 0