work adding import functionality using django-import-export

dev-requirements.txt

@@ -6,4 +6,4 @@ django-debug-toolbar
 sphinx
 wheel
 pre-commit
-wagtail-factories
+wagtail-factories

mep/people/admin.py

@@ -9,7 +9,6 @@
 from django.utils.timezone import now
 from tabular_export.admin import export_to_csv_response
 from viapy.widgets import ViafWidget
-
 from mep.accounts.admin import AddressInline
 from mep.common.admin import (
     CollapsedTabularInline,
@@ -27,6 +26,43 @@
     Relationship,
     RelationshipType,
 )
+from import_export.admin import (
+    ImportExportModelAdmin,
+)
+from import_export.resources import ModelResource
+from import_export.widgets import ManyToManyWidget, Widget
+from import_export.fields import Field
+from parasolr.django.signals import IndexableSignalHandler
+
+PERSON_IMPORT_COLUMNS = (
+    'slug',
+    'gender',
+    'nationalities'
+)
+
+PERSON_IMPORT_EXPORT_COLUMNS = (
+    'slug',
+    'name',
+    'birth_year',
+    'death_year',
+    'gender',
+    'nationalities',
+    'notes',
+    'start_year',
+    'end_year',
+    'mep_id',
+    'sort_name',
+    'viaf_id',
+    'is_organization',
+    'verified',
+    'title',
+    'profession',
+    'relations',
+    'public_notes',
+    'locations',
+    'updated_at',
+    'id',
+)
 
 
 class InfoURLInline(CollapsibleTabularInline):
@@ -82,7 +118,6 @@ class Meta:
             )
         }
 
-
 class CountryAdmin(admin.ModelAdmin):
     form = CountryAdminForm
     list_display = ("name", "geonames_id", "code")
@@ -436,8 +471,106 @@ class Media:
         ]
 
 
+class ExportPersonResource(ModelResource):
+    class Meta:
+        model = Person
+        fields = PERSON_IMPORT_EXPORT_COLUMNS
+        export_order = PERSON_IMPORT_EXPORT_COLUMNS
+
+
+class PersonResource(ModelResource):
+    def __init__(self,*x,**y):
+        super().__init__(*x,**y)
+        # list to contain updated objects for batch indexing at end
+        self.objects_to_index = []
+
+    def before_import(self, dataset, *args, **kwargs):
+        # lower and camel_case headers
+        dataset.headers = [x.lower().replace(' ','_') for x in dataset.headers]
+
+        # turn off indexing temporarily
+        IndexableSignalHandler.disconnect()
+
+        # turn off viaf lookups
+        settings.SKIP_VIAF_LOOKUP = True
+
+    def before_import_row(self, row, **kwargs):
+        """
+        Called on an OrderedDictionary of row attributes.
+        Opportunity to do quick string formatting as a
+        principle of charity to annotators before passing
+        values into django-import-export lookup logic.
+        """
+        # just make sure nation has no string padding
+        row['nation'] = str(row.get('nation')).strip()
+
+        # gender to one char
+        gstr = str(row.get('gender')).strip()
+        row['gender']=gstr[0].upper() if gstr else ''
+
+    def after_save_instance(self, instance, using_transactions, dry_run):
+        """
+        Called when an instance either was or would be saved (depending on dry_run)
+        """
+        self.objects_to_index.append(instance)
+        return super().after_save_instance(instance, using_transactions, dry_run)
+
+    def after_import(self, dataset, result, using_transactions, dry_run, **kwargs):
+        """
+        Called after importing, twice: once with dry_run==True (preview), 
+        once dry_run==False. We report how many objects were updated and need to be indexed.
+        We only do so when dry_run is False.
+        """
+        # run parent method
+        super().after_import(dataset, result, using_transactions, dry_run, **kwargs)
+
+        # report how many need indexing
+        print(f'indexing {len(self.objects_to_index)} objects, dry_run = {dry_run}')        
+
+        # only continue if not a dry run
+        if not dry_run:
+            # re-enable indexing
+            IndexableSignalHandler.connect()
+
+            # index objects
+            if self.objects_to_index:
+                Person.index_items(self.objects_to_index)
+
+        # turn viaf lookups back on
+        settings.SKIP_VIAF_LOOKUP = False
+
+
+
+    # only customized fields need specifying here
+    nationalities = Field(
+        column_name='nationalities',
+        attribute='nationalities',
+        widget=ManyToManyWidget(Country, field='name', separator=';')
+    )
+
+    class Meta:
+        model = Person
+        fields = PERSON_IMPORT_COLUMNS
+        import_id_fields = ('slug',)
+        export_order = PERSON_IMPORT_COLUMNS
+        skip_unchanged = True
+        report_skipped = True
+
+
+class PersonAdminImportExport(PersonAdmin, ImportExportModelAdmin):
+    resource_class = PersonResource
+
+
+    def get_export_resource_class(self):
+        """
+        Specifies the resource class to use for exporting,
+        so that separate fields can be exported than those imported
+        """
+        return ExportPersonResource
+
+
 # enable default admin to see imported data
-admin.site.register(Person, PersonAdmin)
+admin.site.register(Person, PersonAdminImportExport)
 admin.site.register(Country, CountryAdmin)
 admin.site.register(Location, LocationAdmin)
 admin.site.register(Profession, NamedNotableAdmin)

mep/people/models.py

@@ -1,7 +1,7 @@
 import datetime
 import logging
 from string import punctuation
-
+from django.conf import settings
 from django.apps import apps
 from django.contrib.contenttypes.fields import GenericRelation
 from django.core.exceptions import MultipleObjectsReturned
@@ -533,7 +533,7 @@ def save(self, *args, **kwargs):
         """Adds birth and death dates if they aren't already set
         and there's a viaf id for the record"""
 
-        if self.viaf_id and not self.birth_year and not self.death_year:
+        if not getattr(settings,'SKIP_VIAF_LOOKUP',False) and self.viaf_id and not self.birth_year and not self.death_year:
             self.set_birth_death_years()
 
         # if slug has changed, save the old one as a past slug

mep/people/tests/test_admin.py

@@ -1,22 +1,39 @@
 from unittest.mock import Mock, patch
-
+from io import StringIO
+import csv
+import os
+import random
+import tempfile
 from django.contrib import admin
-from datetime import date
-
+from datetime import date, datetime
 from django.http import HttpResponseRedirect
-from django.test import TestCase
+from django.test import TestCase, Client
 from django.urls import reverse
 from django.utils.timezone import now
+from django.apps import apps
 
 from mep.accounts.models import Account, Subscription
 from mep.books.models import Creator, CreatorType, Work
-from mep.people.admin import PersonAdmin, PersonTypeListFilter
-from mep.people.models import Person, PastPersonSlug
-
+from mep.people.admin import PersonAdmin, PersonTypeListFilter, PersonAdminImportExport, PERSON_IMPORT_EXPORT_COLUMNS, ExportPersonResource
+from mep.people.models import Person, PastPersonSlug, Country
+from django.conf import settings
 
 class TestPersonAdmin(TestCase):
     fixtures = ["sample_people"]
 
+    def setUp(self):
+        User = apps.get_model("auth", "User")
+        # script user needed for log entry logic
+        # store the password to login later
+        password = 'adminpass' 
+        self.admin_user = User.objects.create_superuser('admin', 'admin@admin.com', password)
+        self.client = Client()
+        # You'll need to log him in before you can send requests through the client
+        self.client.login(username=self.admin_user.username, password=password)
+        self.url_person_import = '/admin/people/person/import/'
+        self.url_person_process_import = '/admin/people/person/process_import/'
+        self.url_person_export = '/admin/people/person/export/'
+
     def test_merge_people(self):
         mockrequest = Mock()
         test_ids = ["5", "33", "101"]
@@ -99,6 +116,129 @@ def test_export_csv(self, mock_export_to_csv_response):
             # or title case for property with no verbose name
             assert "Is Creator" in headers
 
+
+    def _djangoimportexport_do_export_post(self, file_format=0):
+        response = self.client.post(self.url_person_export, {'file_format':str(file_format)})
+        return response
+
+
+    def test_djangoimportexport_export(self):
+        ### test can get page
+        response = self.client.get(self.url_person_export)
+        self.assertEqual(response.status_code, 200)
+
+        ### test can post to page and get csv data back
+        date_str = datetime.now().strftime("%Y-%m-%d")
+        response = self._djangoimportexport_do_export_post(file_format=0) # csv
+
+        # test response
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(response.has_header("Content-Disposition"))
+        self.assertEqual(response["Content-Type"], "text/csv")
+        self.assertEqual(
+            response["Content-Disposition"],
+            'attachment; filename="Person-{}.csv"'.format(date_str),
+        )
+
+        # test csv as binary string response
+        lines = response.content.splitlines()
+        assert len(lines)>0, 'no header returned'
+        self.assertEqual(
+            ','.join(PERSON_IMPORT_EXPORT_COLUMNS).encode(),
+            lines[0],
+        )
+
+        # test csv via csv reader
+        f = StringIO(response.content.decode())
+        reader = csv.DictReader(f, delimiter=',')
+        rows = list(reader)
+        persons = Person.objects.all()
+
+        # test num lines, should be a row per person
+        assert len(rows) == len(persons)
+
+
+        # test values by row
+        person_admin = PersonAdminImportExport(model=Person, admin_site=admin.site)
+        export_class = person_admin.get_export_resource_class()
+        exporter = export_class()
+
+        def getstr(person,attr,default=''):
+            field = exporter.fields[attr]
+            res = exporter.export_field(field, person)
+            return str(res) if res or res==0 else default
+
+        for person,row in zip(persons,rows):
+            for attr in PERSON_IMPORT_EXPORT_COLUMNS:
+                self.assertEquals(getstr(person,attr), row[attr])
+
+
+    def _djangoimportexport_do_import_post(
+        self, 
+        url, 
+        filename, 
+        input_format=0,
+        follow=False
+    ):
+        with open(filename, "rb") as f:
+            data = {
+                "input_format": str(input_format),
+                "import_file": f,
+            }
+            response = self.client.post(url, data, follow=follow)
+        return response
+
+    def test_djangoimportexport_import(self):
+        ### test can get page
+        response = self.client.get(self.url_person_import)
+        self.assertEqual(response.status_code, 200)
+        self.assertTemplateUsed(response, "admin/import_export/import.html")
+        self.assertContains(response, 'form action=""')
+
+        tmpfn = 'persons.csv'
+        ## test import with changed data
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            csv_filename = os.path.join(tmpdir,tmpfn)
+            # quick export
+            response = self._djangoimportexport_do_export_post()
+            # modify
+            f = StringIO(response.content.decode())
+            reader = csv.DictReader(f, delimiter=',')
+            rows = list(reader)
+            countries = [c.name for c in Country.objects.all()]
+            for row in rows:
+                row['gender'] = random.choice([x for x,y in Person.GENDER_CHOICES if x!=row['gender']])
+                row['nationalities'] = random.choice([x for x in countries if x!=row['nationalities']])
+            # save
+            with open(csv_filename,'w') as of:
+                writer = csv.DictWriter(of, fieldnames=reader.fieldnames)
+                writer.writeheader()
+                writer.writerows(rows)
+
+            # now import
+            response = self._djangoimportexport_do_import_post(self.url_person_import, csv_filename)
+            self.assertEqual(response.status_code, 200)
+            self.assertIn("result", response.context)
+            self.assertFalse(response.context["result"].has_errors())
+            self.assertIn("confirm_form", response.context)
+            confirm_form = response.context["confirm_form"]
+
+            data = confirm_form.initial
+            self.assertEqual(data["original_file_name"], tmpfn)
+            response = self.client.post(self.url_person_process_import, data, follow=True)
+            self.assertEqual(response.status_code, 200)
+            self.assertContains(
+                response,
+                ("Import finished, with {} new and {} updated {}.").format(
+                    0, len(rows), Person._meta.verbose_name_plural
+                ),
+            )
+
+            assert response.content.count(b'<tr class="grp-row') == len(rows)
+
+
+
     def test_past_slugs_list(self):
         person_admin = PersonAdmin(model=Person, admin_site=admin.site)
         person = Person.objects.order_by("id").first()
@@ -170,3 +310,10 @@ def test_queryset(self):
         assert foo in qs
         assert not engelbert in qs
         assert not humperdinck in qs
+
+
+# New tests:
+# `before_import_row
+# import formatting methods for gender and nation (you'll need to adjust the nesting)
+# get_import_fields
+# Person.save optional behavior (skip viaf lookup)

mep/settings.py

@@ -69,6 +69,7 @@
     "mep.books",
     "mep.footnotes",
     "mep.pages",
+    "import_export"
 ]
 
 MIDDLEWARE = [
@@ -323,4 +324,6 @@
     }
 }
 
-WAGTAILADMIN_BASE_URL = "https://shakespeareandco.princeton.edu/cms/"
+WAGTAILADMIN_BASE_URL = "https://shakespeareandco.princeton.edu/cms/"
+
+SKIP_VIAF_LOOKUP = False