Add admin view to Import Tabular Items from a CSV file (#3)

* Add admin view to Import Tabular Items from a CSV file

* Lint code

Author: willemarcel

.gitignore

@@ -105,3 +105,6 @@ venv.bak/
 
 # mypy
 .mypy_cache/
+
+# macos
+.DS_Store

vbos/config/common.py

@@ -88,7 +88,7 @@ class Common(Configuration):
     TEMPLATES = [
         {
             "BACKEND": "django.template.backends.django.DjangoTemplates",
-            "DIRS": STATICFILES_DIRS,
+            "DIRS": [os.path.join(BASE_DIR, "templates")],
             "APP_DIRS": True,
             "OPTIONS": {
                 "context_processors": [

vbos/datasets/admin.py

@@ -1,11 +1,19 @@
+import csv
+from io import TextIOWrapper
+
 from django.contrib.gis import admin
+from django.contrib import messages
+from django.shortcuts import render, redirect, reverse
+from django.urls import path
+
 from .models import (
     RasterDataset,
     TabularDataset,
     TabularItem,
     VectorDataset,
     VectorItem,
 )
+from .forms import CSVUploadForm
 
 
 @admin.register(RasterDataset)
@@ -30,4 +38,73 @@ class TabularDatasetAdmin(admin.ModelAdmin):
 
 @admin.register(TabularItem)
 class TabularItemAdmin(admin.GISModelAdmin):
-    list_display = ["id", "data"]
+    list_display = ["id", "dataset", "data"]
+
+    def get_urls(self):
+        urls = super().get_urls()
+        custom_urls = [
+            path(
+                "upload-file/",
+                self.admin_site.admin_view(self.import_file),
+                name="datasets_tabularitem_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_tabularitem_import_file")
+        return super().changelist_view(request, extra_context=extra_context)
+
+    def import_file(self, request):
+        if request.method == "POST":
+            form = CSVUploadForm(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(".csv"):
+                    messages.error(request, "Please upload a CSV file")
+                    return redirect("admin:datasets_tabularitem_import_file")
+
+                try:
+                    # Read and process the CSV
+                    decoded_file = TextIOWrapper(uploaded_file.file, encoding="utf-8")
+                    reader = csv.DictReader(decoded_file)
+
+                    created_count = 0
+                    error_count = 0
+
+                    for row in reader:  # start=2 to account for header row
+                        try:
+                            TabularItem.objects.create(
+                                dataset=form.cleaned_data["dataset"], data=row
+                            )
+                            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 CSV: {str(e)}")
+
+                return redirect("admin:datasets_tabularitem_import_file")
+        else:
+            form = CSVUploadForm()
+
+        context = {
+            "form": form,
+            "opts": self.model._meta,
+            "title": "Import CSV File",
+        }
+        return render(request, "admin/file_upload.html", context)

vbos/datasets/forms.py

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

vbos/datasets/test/test_admin.py

@@ -0,0 +1,61 @@
+import io
+
+from django.contrib.auth import get_user_model
+from django.test import TestCase, Client
+from django.urls import reverse
+
+from vbos.datasets.models import TabularDataset, TabularItem
+
+
+class TabularItemAdminImportFileTests(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 = TabularDataset.objects.create(name="Test Dataset")
+        self.upload_url = reverse("admin:datasets_tabularitem_import_file")
+
+    def test_change_list_has_link_to_import_file(self):
+        response = self.client.get(reverse("admin:datasets_tabularitem_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 CSV 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 csv")
+        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 CSV file")
+
+    def test_post_valid_csv_creates_items(self):
+        csv_content = "col1,col2\nval1,val2\nval3,val4\n"
+        file_data = io.BytesIO(csv_content.encode("utf-8"))
+        file_data.name = "test.csv"
+        response = self.client.post(
+            self.upload_url,
+            {"file": file_data, "dataset": self.dataset.id},
+            follow=True,
+        )
+        self.assertContains(response, "Successfully created 2 new records")
+        self.assertEqual(TabularItem.objects.count(), 2)
+        ti_1 = TabularItem.objects.first()
+        self.assertEqual(ti_1.dataset.id, self.dataset.id)
+        self.assertEqual(ti_1.data["col1"], "val1")
+        self.assertEqual(ti_1.data["col2"], "val2")
+        ti_2 = TabularItem.objects.last()
+        self.assertEqual(ti_2.dataset.id, self.dataset.id)
+        self.assertEqual(ti_2.data["col1"], "val3")
+        self.assertEqual(ti_2.data["col2"], "val4")

vbos/templates/admin/datasets/tabularitem/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 %}

vbos/templates/admin/file_upload.html

@@ -0,0 +1,40 @@
+<!-- templates/admin/file_upload.html -->
+{% extends "admin/base_site.html" %}
+{% load i18n admin_urls static %}
+{% block content %}
+<div id="content-main">
+    <form method="post" enctype="multipart/form-data">
+        {% csrf_token %}
+
+        <div class="form-row">
+            {{ form.file.label_tag }}
+            {{ form.file }}
+            <div>
+                <p>
+                    {{ form.dataset.label_tag }}
+                </p>
+                <p>
+                    {{ form.dataset }}
+                </p>
+            </div>
+            {% if form.csv_file.errors %}
+                <div class="error">
+                    {{ form.csv_file.errors }}
+                </div>
+            {% endif %}
+        </div>
+
+        <div class="submit-row">
+            <input
+                type="submit"
+                value="{% trans 'Import File' %}"
+                class="default"
+            />
+        </div>
+    </form>
+
+    {% if form.errors %}
+        <div class="errornote">{% trans "Please correct the errors below." %}</div>
+    {% endif %}
+</div>
+{% endblock %}