# (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 submits the file as an async DataTask (app_data.tasks.submit_import), which writes it through load_broker (run_import_batch()) in the worker. Every import is a task, whatever its size (5.4.0; there used to be a row-count threshold, settings.CAVALIBA_MAX_SYNC_IMPORT, below which the write loop ran inside the request) - so progress, abort and the result report are the same UI for a 5-row file and a 500k-row one. Method inventory: import_tool, import_progress, import_abort. Private helpers: _run_and_submit. """ from django.contrib import messages from django.http import Http404, HttpResponseNotAllowed 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 from app_data.pipeline import Pipeline from app_data.schema import Schema from app_data.task_manager import abort_task from app_data.task_ui import own_task, progress_context from app_data.tasks import submit_import from app_home.log import WARNING, log2 _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, then either re-render in place (verify / parse error / invalid form) or submit the import as a task and re-render with its handle, which opens the progress modal.""" context = start_view( request, app="data", view="import_tool", 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)] # always present, so the template (and tests) can read it unconditionally: # only a submitted import sets it, and only then does the modal open context["task_handle"] = None if request.method == "POST": form = ImportForm( request.POST, request.FILES, pipelines=pipelines, schemas=schemas, is_admin=is_data_admin, ) if form.is_valid(): # a submitted import returns its task handle, which the template # turns into the progress modal; every other outcome (verify, parse # error, unsupported type) returns None and just re-renders context["task_handle"] = _run_and_submit(request, aaa, form, is_data_admin) 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_submit(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 submit it as a DataTask. Every import runs as a task, whatever its size: the write loop never runs inside the request, so one progress/abort UI covers all of them. The file is still parsed here rather than only in the worker - that is what reports a broken file (encoding, syntax) immediately instead of turning it into a FAILED task the user has to go open, and it supplies the row count. Returns the task handle (str) when an import was submitted, else None - a verify, a parse error or an unsupported extension all leave the caller to re-render the page with the bound (and so still filled in) form.""" cleaned = form.cleaned_data postfile = cleaned["file"] # only the name travels: the worker resolves the Pipeline itself, and the # form's choices already come from Pipeline.list(is_enabled=True) pipeline_name = cleaned.get("pipeline") or None # force_action/force_schema are only ever present in cleaned_data for an # admin form instance (ImportForm.__init__ deletes both 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 # sync_mode is a common option (checked by default) - Instance.save() # reads aaa["sync_mode"] 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")) # loader-native 1-based inclusive row range: first defaults to 1 (no skip), # last to 0 (no limit) - the loaders' own defaults. first = cleaned.get("first") or 1 last = cleaned.get("last") or 0 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}") log2.fail("import.file", reason="parse_error", message=str(err), extra={"file": postfile.name}) # fmt: skip return None submit = request.POST.get("submit") if submit == "verify": # name the delimiter when the form detected it: Check is where you go # to confirm a file reads correctly, and a plausible-looking row count # on the wrong delimiter is exactly what it should help you catch note = "" if form.detected_separator: note = " - " + _("delimiter detected: {sep}").format(sep=form.detected_separator) messages.add_message( request, messages.SUCCESS, _("Check ok") + f" ({len(datalist)} rows)" + note ) log2.debug("import check ok: %s rows", len(datalist)) return None if submit == "import": 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"], display_name=postfile.name, ) log2.ok("import.submit", count=len(datalist), target=("task", handle)) return handle return None # ------------------------------------------------------------------------- # IMPORT TOOL - task modal : progress / abort # ------------------------------------------------------------------------- def import_progress(request, handle): """htmx partial: one render of the import progress modal body. Re-requests itself every 2s while the task is live and stops polling once it is terminal. Same shape as export_progress - the ownership check and the figures come from the shared app_data/task_ui.py.""" context = start_view( request, app="data", view="import_progress", perm="p_data_import", noauthz="app_home:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] task = own_task(handle, aaa) if not task: raise Http404 progress_context(context, task) return render(request, "app_data/import_progress.html", context) def import_abort(request, handle): """POST from the modal's Abort button: request the task stop, then re-render the modal body. The worker notices on its next batch - so an abort stops the remaining rows, it does not undo the ones already written (unlike an aborted export, whose partial file is simply discarded).""" context = start_view( request, app="data", view="import_abort", perm="p_data_import", noauthz="app_home:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] task = own_task(handle, aaa) if not task: raise Http404 if request.method != "POST": return HttpResponseNotAllowed(["POST"]) stopped = abort_task(handle, aaa=aaa) if stopped: log2.ok("import.abort", target=("task", handle)) else: log2.fail("import.abort", reason="not_running", target=("task", handle), level=WARNING) task.refresh_from_db() progress_context(context, task) return render(request, "app_data/import_progress.html", context)