UUID bulk update

docs/source/_ext/djangodocs.py

@@ -200,13 +200,13 @@ def finish(self):
         templatebuiltins = {
             "ttags": [
                 n
-                for ((t, n), (l, a)) in xrefs.items()
-                if t == "templatetag" and l == "ref/templates/builtins"
+                for ((t, n), (l, a)) in xrefs.items()  # noqa
+                if t == "templatetag" and l == "ref/templates/builtins"  # noqa
             ],
             "tfilters": [
                 n
-                for ((t, n), (l, a)) in xrefs.items()
-                if t == "templatefilter" and l == "ref/templates/builtins"
+                for ((t, n), (l, a)) in xrefs.items()  # noqa
+                if t == "templatefilter" and l == "ref/templates/builtins"  # noqa
             ],
         }
         outfilename = os.path.join(self.outdir, "templatebuiltins.js")

src/adminactions/bulk_update.py

@@ -10,6 +10,7 @@
 from django.core.exceptions import ValidationError
 from django.core.files.utils import FileProxyMixin
 from django.core.validators import FileExtensionValidator
+from django.db import models
 from django.db.transaction import atomic
 from django.forms import Media
 from django.http import HttpResponseRedirect
@@ -260,13 +261,12 @@
 
         if header:
             reader = csv.DictReader(codecs.iterdecode(f, "utf-8"), **(csv_options or {}))
-            for k, v in mapping.items():
+            for _k, v in mapping.items():
                 if v not in reader.fieldnames:
                     raise ValidationError(_("%s column is not present in the file") % v)
         else:
             reader = csv.reader(codecs.iterdecode(f, "utf-8"), **(csv_options or {}))
             mapping = {k: int(v) - 1 for k, v in mapping.items()}
-
         reverse = {v: k for k, v in mapping.items()}
         with atomic():
             for i, row in enumerate(reader, 1):
@@ -278,7 +278,21 @@
                         for colname, value in row.items():
                             field = reverse[colname]
                             if field not in indexes:
-                                changes[field] = [getattr(obj, field), value]
+                                model_field = queryset.model._meta.get_field(field)
+                                if model_field.is_relation and model_field.many_to_one:
+                                    related_model = model_field.related_model
+                                    related_field_name = (
+                                        model_field.to_fields[0] if model_field.to_fields else "pk"
+                                    )
+                                    related_field = related_model._meta.get_field(related_field_name)
+
+                                    if isinstance(related_field, models.UUIDField):
+                                        try:
+                                            value = related_model.objects.get(**{related_field_name: value})
+                                        except related_model.DoesNotExist:
+                                            raise ValidationError(
+                                                f"No instance of {related_model._meta.verbose_name} found with {related_field_name} = {value}"
+                                            )
                                 setattr(obj, field, value)
                     else:
                         for i, value in enumerate(row):

tests/demo/migrations/0001_initial.py

@@ -1,5 +1,7 @@
 # Generated by Django 2.0.1 on 2018-01-29 00:00
 
+import uuid
+
 import demo.models
 import django.db.models.deletion
 from django.conf import settings
@@ -39,6 +41,10 @@ class Migration(migrations.Migration):
                 ("generic_ip", models.GenericIPAddressField()),
                 ("url", models.URLField()),
                 ("text", models.TextField()),
+                (
+                    "uuid",
+                    models.UUIDField(default=uuid.uuid4, editable=False, unique=True),
+                ),
                 ("unique", models.CharField(max_length=255, unique=True)),
                 ("nullable", models.CharField(max_length=255, null=True)),
                 ("blank", models.CharField(blank=True, max_length=255, null=True)),
@@ -82,6 +88,29 @@ class Migration(migrations.Migration):
                 ),
             ],
         ),
+        migrations.CreateModel(
+            name="DemoRelated",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "demo",
+                    models.ForeignKey(
+                        on_delete=models.deletion.CASCADE,
+                        to="demo.DemoModel",
+                        related_name="related",
+                        to_field="uuid",
+                    ),
+                ),
+            ],
+        ),
         migrations.CreateModel(
             name="UserDetail",
             fields=[

tests/demo/models.py

@@ -1,3 +1,5 @@
+import uuid
+
 from admin_extra_urls.api import button
 from admin_extra_urls.mixins import ExtraUrlMixin
 from django.contrib.admin import ModelAdmin, site
@@ -29,6 +31,7 @@ class DemoModel(models.Model):
     generic_ip = models.GenericIPAddressField()
     url = models.URLField()
     text = models.TextField()
+    uuid = models.UUIDField(default=uuid.uuid4, editable=False, unique=True)
 
     unique = models.CharField(max_length=255, unique=True)
     nullable = models.CharField(max_length=255, null=True)
@@ -59,6 +62,13 @@ class Meta:
         app_label = "demo"
 
 
+class DemoRelated(models.Model):
+    demo = models.ForeignKey(DemoModel, on_delete=models.CASCADE, related_name="related", to_field="uuid")
+
+    class Meta:
+        app_label = "demo"
+
+
 class UserDetailModelAdmin(ExtraUrlMixin, ModelAdmin):
     list_display = [f.name for f in UserDetail._meta.fields]
 
@@ -88,6 +98,10 @@ class DemoOneToOneAdmin(ExtraUrlMixin, AdminActionPermMixin, ModelAdmin):
     pass
 
 
+class DemoRelatedAdmin(ExtraUrlMixin, AdminActionPermMixin, ModelAdmin):
+    pass
+
+
 class TestMassUpdateForm(MassUpdateForm):
     pass
 
@@ -98,4 +112,5 @@ class DemoModelMassUpdateForm(MassUpdateForm):
 
 site.register(DemoModel, DemoModelAdmin)
 site.register(DemoOneToOne, DemoOneToOneAdmin)
+site.register(DemoRelated, DemoRelatedAdmin)
 site.register(UserDetail, UserDetailModelAdmin)

tests/test_bulk_update.py

@@ -1,7 +1,7 @@
 import csv
 from pathlib import Path
 
-from demo.models import DemoModel
+from demo.models import DemoModel, DemoOneToOne, DemoRelated
 from django.conf import settings
 from django.contrib.auth.models import User
 from django.test import TestCase
@@ -23,6 +23,9 @@ class BulkUpdate(SelectRowsMixin, CheckSignalsMixin, WebTestMixin):
     csrf_checks = True
 
     _selected_rows = [0, 1]
+    _selectedr_rows = [
+        0,
+    ]
 
     action_name = "bulk_update"
     sender_model = DemoModel
@@ -64,6 +67,37 @@ def _run_action(self, steps=2, **kwargs):
                 res = res.forms["bulk-update"].submit("apply")
         return res
 
+    def _run_action_related_model(self, steps=2, **kwargs):
+        selected_rows = kwargs.pop("selected_rows", self._selectedr_rows)
+        select_across = kwargs.pop("select_across", False)
+        with user_grant_permission(
+            self.user,
+            ["demo.change_demorelated", "demo.adminactions_bulkupdate_demorelated"],
+        ):
+            res = self.app.get("/", user="user", auto_follow=False)
+            res = res.click("Demo related")
+            # print(res)
+            if steps >= 1:
+                form = res.forms["changelist-form"]
+                form["action"] = "bulk_update"
+                form["select_across"] = select_across
+                self._select_rows(form, selected_rows)
+                res = form.submit()
+            if steps >= 2:
+                res.forms["bulk-update"]["_file"] = Upload(
+                    str(Path(__file__).parent / "related_model_bulk_update.csv")
+                )
+                res.forms["bulk-update"]["fld-id"] = "id"
+                res.forms["bulk-update"]["fld-index_field"] = ["id"]
+                res.forms["bulk-update"]["fld-demo"] = "demo_uuid"
+                res.forms["bulk-update"]["csv-delimiter"] = ","
+                res.forms["bulk-update"]["csv-quoting"] = csv.QUOTE_NONE
+
+                for k, v in kwargs.items():
+                    res.forms["bulk-update"][k] = v
+                res = res.forms["bulk-update"].submit("apply")
+        return res
+
     def test_simulate(self):
         res = self._run_action(
             **{
@@ -184,6 +218,47 @@ def test_wrong_mapping(self):
         messages = [m.message for m in list(res.context["messages"])]
         assert messages[0] == "['miss column is not present in the file']"
 
+    def test_bulk_update_with_one_to_one_field(self):
+        demo_model_instance = G(DemoModel, char="InitialValue", integer=123)
+        demo_one_to_one_instance = G(DemoOneToOne, demo=demo_model_instance)
+        csv_data = f"pk,one_to_one_id\n{demo_model_instance.pk},{demo_one_to_one_instance.pk}"
+        res = self._run_action(
+            **{
+                "_file": Upload(
+                    "data.csv",
+                    csv_data.encode(),
+                    "text/csv",
+                ),
+                "fld-onetoone": "one_to_one_id",
+            }
+        )
+        self.assertTrue(
+            DemoModel.objects.filter(pk=demo_model_instance.pk, onetoone=demo_one_to_one_instance).exists()
+        )
+        self.assertEqual(res.status_code, 200)
+
+    def test_bulk_update_with_foreign_key(self):
+        demo_model_instance = G(DemoModel, char="InitialValue", integer=123)
+        demo_related_instance = G(DemoRelated, demo=demo_model_instance)
+        new_demo_model_instance = G(DemoModel, char="NewValue", integer=456)
+        csv_data = f"id,demo_uuid\n{demo_related_instance.pk},{new_demo_model_instance.uuid}"
+
+        res = self._run_action_related_model(
+            **{
+                "_file": Upload(
+                    "data.csv",
+                    csv_data.encode(),
+                    "text/csv",
+                ),
+                "fld-demo": "demo_uuid",
+            }
+        )
+
+        self.assertTrue(
+            DemoRelated.objects.filter(pk=demo_related_instance.pk, demo=new_demo_model_instance).exists()
+        )
+        self.assertEqual(res.status_code, 200)
+
 
 class BulkUpdateMemoryFileUploadHandlerTest(BulkUpdate, TestCase):
     handler = "django.core.files.uploadhandler.MemoryFileUploadHandler"

tests/test_permissions.py

@@ -61,7 +61,7 @@ def test_permission_needed(app, admin, demomodels, action):
 
 @pytest.mark.django_db()
 def test_permissions(admin):
-    assert Permission.objects.filter(codename__startswith="adminactions").count() == 63
+    assert Permission.objects.filter(codename__startswith="adminactions").count() == 70
 
     with user_grant_permission(admin, ["demo.adminactions_export_demomodel"]):
         assert admin.get_all_permissions() == set(["demo.adminactions_export_demomodel"])