add the admin flow for import/export

src/open_inwoner/openzaak/admin.py

@@ -1,14 +1,25 @@
+import datetime
+import logging
+
 from django.contrib import admin, messages
 from django.core.exceptions import ValidationError
 from django.db.models import BooleanField, Count, ExpressionWrapper, Q
-from django.forms import ModelForm, Textarea
+from django.forms import Form, ModelForm, Textarea
 from django.forms.models import BaseInlineFormSet
+from django.http import HttpResponseRedirect, StreamingHttpResponse
+from django.template.response import TemplateResponse
+from django.urls import path, reverse
+from django.utils.safestring import mark_safe
 from django.utils.translation import gettext_lazy as _, ngettext
 
 from import_export.admin import ImportExportMixin
+from privates.storages import PrivateMediaFileSystemStorage
 from solo.admin import SingletonModelAdmin
 
 from open_inwoner.ckeditor5.widgets import CKEditorWidget
+from open_inwoner.openzaak.import_export import ZGWImportExportService
+from open_inwoner.openzaak.tasks import process_zgw_import_file
+from open_inwoner.utils.forms import LimitedUploadFileField
 
 from .models import (
     CatalogusConfig,
@@ -22,6 +33,8 @@
     ZGWApiGroupConfig,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class ZGWApiGroupConfig(admin.StackedInline):
     model = ZGWApiGroupConfig
@@ -67,8 +80,24 @@ class OpenZaakConfigAdmin(SingletonModelAdmin):
     )
 
 
+class ImportZGWDumpForm(Form):
+    zgw_export_file = LimitedUploadFileField(
+        label=_("ZGW export bestand"),
+        help_text=_(
+            "Upload your file previously generated by the export function. Max size is 25MB"
+        ),
+        allow_empty_file=False,
+        required=True,
+        max_upload_size=1024**2 * 25,
+        min_upload_size=1,
+        allowed_mime_types=["application/json", "text/plain"],
+    )
+
+
 @admin.register(CatalogusConfig)
 class CatalogusConfigAdmin(admin.ModelAdmin):
+    change_list_template = "admin/catalogusconfig_change_list.html"
+
     list_display = [
         "domein",
         "rsin",
@@ -90,6 +119,88 @@ class CatalogusConfigAdmin(admin.ModelAdmin):
     ordering = ("domein", "rsin")
     list_filter = ("service",)
 
+    actions = ["export_catalogus_configs"]
+
+    zgw_import_storage = PrivateMediaFileSystemStorage()
+    zgw_import_export_service = ZGWImportExportService()
+
+    def get_urls(self):
+        urls = super().get_urls()
+        custom_urls = [
+            path(
+                "import-catalogus-dump/",
+                self.admin_site.admin_view(self.process_file_view),
+                name="upload_zgw_import_file",
+            ),
+        ]
+        return custom_urls + urls
+
+    @admin.action(description=_("Exporteer naar json export"))
+    def export_catalogus_configs(modeladmin, request, queryset):
+        response = StreamingHttpResponse(
+            modeladmin.zgw_import_export_service.export_catalogus_configs_jsonl(
+                queryset
+            ),
+            content_type="application/json",
+        )
+        response[
+            "Content-Disposition"
+        ] = 'attachment; filename="zgw-catalogi-export.json"'
+        return response
+
+    def process_file_view(self, request):
+        form = None
+
+        match request.method:
+            case "POST":
+                form = ImportZGWDumpForm(request.POST, request.FILES)
+                if form.is_valid():
+                    uploaded_file = request.FILES["zgw_export_file"]
+                    timestamp = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
+                    target_file_name = f"zgw_import_dump_{timestamp}.jsonl"
+                    self.zgw_import_storage.save(target_file_name, uploaded_file)
+
+                    try:
+                        task = process_zgw_import_file.delay(filename=target_file_name)
+
+                        celery_monitor_url = (
+                            reverse(
+                                "admin:celery_monitor_taskstate_changelist",
+                            )
+                            + "?name=open_inwoner.openzaak.tasks.process_zgw_import_file"
+                        )
+
+                        message = _(
+                            'Your import has been successfully received and is being processed in the background. You can <a href="%(task_url)s">monitor the task\'s progress here</a>. Note that it may take a few seconds for your task to appear, you might have to refresh the page a few times.'
+                            % {"task_url": celery_monitor_url}
+                        )
+
+                        self.message_user(
+                            request,
+                            mark_safe(message),
+                            messages.SUCCESS,
+                        )
+                        return HttpResponseRedirect(
+                            reverse(
+                                "admin:openzaak_catalogusconfig_changelist",
+                            )
+                        )
+                    except Exception as e:
+                        logger.exception("Unable to process ZGW import")
+                        self.message_user(
+                            request, f"Error processing file: {str(e)}", messages.ERROR
+                        )
+                else:
+                    self.message_user(
+                        request, _("No file was uploaded"), messages.WARNING
+                    )
+            case _:
+                form = ImportZGWDumpForm()
+
+        return TemplateResponse(
+            request, "admin/import_zgw_dump_form.html", {"form": form}
+        )
+
 
 class HasDocNotifyListFilter(admin.SimpleListFilter):
     title = _("notify document attachment")

src/open_inwoner/openzaak/tasks.py

@@ -2,7 +2,10 @@
 
 from django.core.management import call_command
 
+from privates.storages import PrivateMediaFileSystemStorage
+
 from open_inwoner.celery import app
+from open_inwoner.openzaak.import_export import ZGWImportExportService
 
 logger = logging.getLogger(__name__)
 
@@ -14,3 +17,14 @@ def import_zgw_data():
     call_command("zgw_import_data")
 
     logger.info("finished import_zgw_data() task")
+
+
+@app.task
+def process_zgw_import_file(filename: str):
+    service = ZGWImportExportService()
+    storage = PrivateMediaFileSystemStorage()
+
+    try:
+        return service.import_catalogus_config(filename, storage)
+    finally:
+        storage.delete(filename)

src/open_inwoner/openzaak/tests/test_admin.py

@@ -1,12 +1,24 @@
+import json
+from unittest import mock
+
+from django.test import override_settings
 from django.urls import reverse
 from django.utils.translation import gettext_lazy as _
 
+import freezegun
 from django_webtest import WebTest
 from maykin_2fa.test import disable_admin_mfa
+from privates.storages import PrivateMediaFileSystemStorage
+from webtest import Upload
 
 from open_inwoner.accounts.tests.factories import UserFactory
 
-from .factories import ZaakTypeConfigFactory, ZaakTypeInformatieObjectTypeConfigFactory
+from .factories import (
+    CatalogusConfigFactory,
+    ServiceFactory,
+    ZaakTypeConfigFactory,
+    ZaakTypeInformatieObjectTypeConfigFactory,
+)
 
 
 @disable_admin_mfa()
@@ -109,3 +121,78 @@ def test_both_can_be_disabled(self):
         )
         self.assertFalse(self.ztc.document_upload_enabled)
         self.assertFalse(self.ztiotc.document_upload_enabled)
+
+
+@disable_admin_mfa()
+@override_settings(CELERY_TASK_ALWAYS_EAGER=True)
+class TestCatalogusConfigExportAdmin(WebTest):
+    csrf_checks = False
+
+    def setUp(self):
+        self.user = UserFactory(is_superuser=True, is_staff=True)
+        self.service = ServiceFactory(slug="service-1")
+        self.catalogus = CatalogusConfigFactory(
+            url="https://foo.maykinmedia.nl",
+            domein="DM",
+            rsin="123456789",
+            service=self.service,
+        )
+
+    def test_export_action_returns_json_response(self):
+        response = self.app.post(
+            reverse(
+                "admin:openzaak_catalogusconfig_changelist",
+            ),
+            {
+                "action": "export_catalogus_configs",
+                "_selected_action": [self.catalogus.id],
+            },
+            user=self.user,
+        )
+
+        self.assertEqual(
+            response.content_disposition,
+            'attachment; filename="zgw-catalogi-export.json"',
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            [json.loads(row) for row in response.text.split("\n")[:-1]],
+            [
+                {
+                    "model": "openzaak.catalogusconfig",
+                    "fields": {
+                        "url": "https://foo.maykinmedia.nl",
+                        "domein": "DM",
+                        "rsin": "123456789",
+                        "service": ["service-1"],
+                    },
+                },
+            ],
+        )
+
+    @mock.patch(
+        "open_inwoner.openzaak.import_export.ZGWImportExportService.import_catalogus_config"
+    )
+    def test_actual_import_flow(self, m) -> None:
+        form = self.app.get(
+            reverse(
+                "admin:upload_zgw_import_file",
+            ),
+            user=self.user,
+        ).form
+        form["zgw_export_file"] = Upload("dump.json", b"foobar", "application/json")
+
+        with freezegun.freeze_time("2024-08-14 17:50:01"):
+            response = form.submit().follow()
+
+        self.assertEqual(m.call_count, 1)
+        self.assertEqual(m.call_args[0][0], "zgw_import_dump_2024-08-14-17-50-01.jsonl")
+        self.assertTrue(isinstance(m.call_args[0][1], PrivateMediaFileSystemStorage))
+
+        messages = [str(msg) for msg in response.context["messages"]]
+        self.assertEqual(
+            messages,
+            [
+                'Your import has been successfully received and is being processed in the background. You can <a href="/admin/celery_monitor/taskstate/?name=open_inwoner.openzaak.tasks.process_zgw_import_file">monitor the task\'s progress here</a>. Note that it may take a few seconds for your task to appear, you might have to refresh the page a few times.'
+            ],
+        )

src/open_inwoner/templates/admin/catalogusconfig_change_list.html

@@ -0,0 +1,11 @@
+{% extends "admin/change_list.html" %}
+{% load i18n admin_urls %}
+
+{% block object-tools-items %}
+{{ block.super }}
+<li>
+    <a href="{% url 'admin:upload_zgw_import_file' %}" class="addlink">
+        {% trans "Import from jsonl file" %}
+    </a>
+</li>
+{% endblock %}

src/open_inwoner/templates/admin/import_zgw_dump_form.html

@@ -0,0 +1,12 @@
+{% extends "admin/base_site.html" %}
+{% load i18n admin_urls %}
+
+{% block content %}
+<div>
+    <form method="post" enctype="multipart/form-data">
+        {% csrf_token %}
+        {{ form.as_div }}
+        <input type="submit" value="Submit">
+    </form>
+</div>
+{% endblock %}