Add admin view to import VectorItem from geojson files

Author: willemarcel

vbos/datasets/admin.py

@@ -1,8 +1,10 @@
 import csv
+import json
 from io import TextIOWrapper
 
 from django.contrib.gis import admin
 from django.contrib import messages
+from django.contrib.gis.geos.geometry import GEOSGeometry
 from django.shortcuts import render, redirect, reverse
 from django.urls import path
 
@@ -13,7 +15,7 @@
     VectorDataset,
     VectorItem,
 )
-from .forms import CSVUploadForm
+from .forms import CSVUploadForm, GeoJSONUploadForm
 
 
 @admin.register(RasterDataset)
@@ -28,7 +30,77 @@ class VectorDatasetAdmin(admin.ModelAdmin):
 
 @admin.register(VectorItem)
 class VectorItemAdmin(admin.GISModelAdmin):
-    list_display = ["id", "metadata"]
+    list_display = ["id", "dataset", "metadata"]
+
+    def get_urls(self):
+        urls = super().get_urls()
+        custom_urls = [
+            path(
+                "upload-file/",
+                self.admin_site.admin_view(self.import_file),
+                name="datasets_vectoritem_import_file",
+            ),
+        ]
+        return custom_urls + urls
+
+    def changelist_view(self, request, extra_context=None):
+        extra_context = extra_context or {}
+        extra_context["upload_file"] = reverse("admin:datasets_vectoritem_import_file")
+        return super().changelist_view(request, extra_context=extra_context)
+
+    def import_file(self, request):
+        if request.method == "POST":
+            form = GeoJSONUploadForm(request.POST, request.FILES)
+            if form.is_valid():
+                uploaded_file = request.FILES["file"]
+
+                # Check if the file is a CSV
+                if not uploaded_file.name.endswith(".geojson"):
+                    messages.error(request, "Please upload a GeoJSON file")
+                    return redirect("admin:datasets_vectoritem_import_file")
+
+                try:
+                    decoded_file = TextIOWrapper(uploaded_file.file, encoding="utf-8")
+                    geojson_content = json.loads(decoded_file.read())
+
+                    created_count = 0
+                    error_count = 0
+
+                    for item in geojson_content["features"]:
+                        try:
+                            VectorItem.objects.create(
+                                dataset=form.cleaned_data["dataset"],
+                                metadata=item["properties"],
+                                geometry=GEOSGeometry(json.dumps(item["geometry"])),
+                            )
+                            created_count += 1
+                        except Exception as e:
+                            print(e)
+                            error_count += 1
+
+                    if created_count > 0:
+                        messages.success(
+                            request, f"Successfully created {created_count} new records"
+                        )
+
+                    if error_count > 0:
+                        messages.warning(
+                            request, f"Failed to create {error_count} items."
+                        )
+
+                except Exception as e:
+                    messages.error(request, f"Error processing GeoJSON: {str(e)}")
+
+                return redirect("admin:datasets_vectoritem_import_file")
+        else:
+            form = GeoJSONUploadForm()
+
+        context = {
+            "form": form,
+            "opts": self.model._meta,
+            "title": "Import GeoJSON File",
+        }
+        return render(request, "admin/file_upload.html", context)
 
 
 @admin.register(TabularDataset)

vbos/datasets/fixtures/test.geojson

@@ -0,0 +1,66 @@
+{
+  "type": "FeatureCollection",
+  "features": [
+    {
+      "type": "Feature",
+      "properties": {
+        "name": "Area 1"
+      },
+      "geometry": {
+        "coordinates": [
+          [
+            [
+              167.59888,
+              -16.28524
+            ],
+            [
+              167.59888,
+              -16.35415
+            ],
+            [
+              167.65928,
+              -16.35415
+            ],
+            [
+              167.65928,
+              -16.28524
+            ],
+            [
+              167.59888,
+              -16.28524
+            ]
+          ]
+        ],
+        "type": "Polygon"
+      }
+    },
+    {
+      "type": "Feature",
+      "properties": {"name": "Line 1", "col": "val"},
+      "geometry": {
+        "coordinates": [
+          [
+            167.471331,
+            -16.249016
+          ],
+          [
+            167.500713,
+            -16.372577
+          ]
+        ],
+        "type": "LineString"
+      }
+    },
+    {
+      "type": "Feature",
+      "properties": {"name": "Point 1", "source": "OpenStreetMap"},
+      "geometry": {
+        "coordinates": [
+          167.72470,
+          -16.39183
+        ],
+        "type": "Point"
+      }
+    }
+  ]
+}

vbos/datasets/forms.py

@@ -1,9 +1,16 @@
 from django import forms
-from .models import TabularDataset
+from .models import TabularDataset, VectorDataset
 
 
 class CSVUploadForm(forms.Form):
     file = forms.FileField(label="File")
     dataset = forms.ModelChoiceField(
         queryset=TabularDataset.objects.all(), empty_label="Select a dataset"
     )
+
+
+class GeoJSONUploadForm(forms.Form):
+    file = forms.FileField(label="File")
+    dataset = forms.ModelChoiceField(
+        queryset=VectorDataset.objects.all(), empty_label="Select a dataset"
+    )

vbos/datasets/models.py

@@ -32,7 +32,7 @@ class VectorItem(models.Model):
     metadata = models.JSONField(default=dict, blank=True, null=True)
 
     def __str__(self):
-        return self.id
+        return f"{self.id}"
 
     class Meta:
         ordering = ["id"]

vbos/datasets/test/test_admin.py

@@ -4,7 +4,7 @@
 from django.test import TestCase, Client
 from django.urls import reverse
 
-from vbos.datasets.models import TabularDataset, TabularItem
+from vbos.datasets.models import TabularDataset, TabularItem, VectorDataset, VectorItem
 
 
 class TabularItemAdminImportFileTests(TestCase):
@@ -59,3 +59,55 @@ def test_post_valid_csv_creates_items(self):
         self.assertEqual(ti_2.dataset.id, self.dataset.id)
         self.assertEqual(ti_2.data["col1"], "val3")
         self.assertEqual(ti_2.data["col2"], "val4")
+
+
+class VectorItemAdminImportFileTests(TestCase):
+    def setUp(self):
+        self.client = Client()
+        self.admin_user = get_user_model().objects.create_superuser(
+            username="admin", password="password", email="[email protected]"
+        )
+        self.client.login(username="admin", password="password")
+        self.dataset = VectorDataset.objects.create(name="Test Dataset")
+        self.upload_url = reverse("admin:datasets_vectoritem_import_file")
+
+    def test_change_list_has_link_to_import_file(self):
+        response = self.client.get(reverse("admin:datasets_vectoritem_changelist"))
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Import File")
+
+    def test_get_import_file_view(self):
+        response = self.client.get(self.upload_url)
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Import GeoJSON File")
+        self.assertContains(response, "Import File")
+        self.assertContains(response, "Dataset")
+        self.assertContains(response, "Test Dataset")
+
+    def test_post_invalid_file_type(self):
+        file_data = io.BytesIO(b"not a geojson")
+        file_data.name = "test.txt"
+        response = self.client.post(
+            self.upload_url,
+            {"file": file_data, "dataset": self.dataset.id},
+            follow=True,
+        )
+        self.assertContains(response, "Please upload a GeoJSON file")
+
+    def test_post_valid_geojson_creates_items(self):
+        geojson_path = "./vbos/datasets/fixtures/test.geojson"
+        with open(geojson_path, "rb") as file_data:
+            response = self.client.post(
+                self.upload_url,
+                {"file": file_data, "dataset": self.dataset.id},
+                follow=True,
+            )
+        self.assertContains(response, "Successfully created 3 new records")
+        self.assertEqual(VectorItem.objects.count(), 3)
+        vi_1, vi_2, vi_3 = VectorItem.objects.all()
+        self.assertEqual(vi_1.dataset.id, self.dataset.id)
+        self.assertEqual(vi_1.metadata["name"], "Area 1")
+        self.assertEqual(vi_2.dataset.id, self.dataset.id)
+        self.assertEqual(vi_2.metadata["name"], "Line 1")
+        self.assertEqual(vi_3.dataset.id, self.dataset.id)
+        self.assertEqual(vi_3.metadata["name"], "Point 1")

vbos/templates/admin/datasets/vectoritem/change_list.html

@@ -0,0 +1,10 @@
+{% extends "admin/change_list.html" %}
+{% load admin_urls %}
+{% block object-tools-items %}
+{{ block.super }}
+<li>
+    <a href="{% url opts|admin_urlname:'import_file' %}" class="addlink">
+        Import File
+    </a>
+</li>
+{% endblock %}