Add default_user or default_change_reason for bulk_create or bulk_upd…

…ate (#653)

* Added possibility to specify a default user for bulk_create_with_history

* format

* added tests

* flake

* added more test coverage

Co-authored-by: Benjamin Mampaey <benjamin.mampaey@oma.be>

AUTHORS.rst

@@ -16,6 +16,7 @@ Authors
 - Alexander Anikeev
 - Amanda Ng (`AmandaCLNg <https://github.com/AmandaCLNg>`_)
 - Ben Lawson (`blawson <https://github.com/blawson>`_)
+- Benjamin Mampaey (`bmampaey <https://github.com/bmampaey>`_)
 - `bradford281 <https://github.com/bradford281>`_
 - Brian Armstrong (`barm <https://github.com/barm>`_)
 - Buddy Lindsey, Jr.

CHANGES.rst

@@ -4,6 +4,7 @@ Changes
 Unreleased
 ------------
 - Added `bulk_update_with_history` utility function (gh-650)
+- Add default user and default change reason to `bulk_create_with_history` and `bulk_update_with_history`
 
 2.9.0 (2020-04-23)
 ------------------

docs/common_issues.rst

@@ -32,17 +32,28 @@ history:
     >>> Poll.history.count()
     1000
 
-If you want to specify a change reason for each record in the bulk create, you
-can add `changeReason` on each instance:
+If you want to specify a change reason or history user for each record in the bulk create,
+you can add `changeReason` or `_history_user` on each instance:
 
 .. code-block:: pycon
 
     >>> for poll in data:
             poll.changeReason = 'reason'
+            poll._history_user = my_user
     >>> objs = bulk_create_with_history(data, Poll, batch_size=500)
     >>> Poll.history.get(id=data[0].id).history_change_reason
     'reason'
 
+You can also specify a default user or default change reason responsible for the change
+(`_history_user` and `changeReason` take precedence).
+
+.. code-block:: pycon
+
+    >>> user = User.objects.create_user("tester", "tester@example.com")
+    >>> objs = bulk_create_with_history(data, Poll, batch_size=500, default_user=user)
+    >>> Poll.history.get(id=data[0].id).history_user == user
+    True
+
 Bulk Updating a Model with History (New)
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Bulk update was introduced with Django 2.2. We can use the utility function
@@ -58,11 +69,11 @@ Bulk update was introduced with Django 2.2. We can use the utility function
     >>> data = [Poll(id=x, question='Question ' + str(x), pub_date=now()) for x in range(1000)]
     >>> objs = bulk_create_with_history(data, Poll, batch_size=500)
     >>> for obj in objs: obj.question = 'Duplicate Questions'
-    >>> bulk_update_with_history(objs, Poll, ['question'], batch_size=500)        
+    >>> bulk_update_with_history(objs, Poll, ['question'], batch_size=500)
     >>> Poll.objects.first().question
     'Duplicate Question'
 
-QuerySet Updates with History (Updated in Django 2.2) 
+QuerySet Updates with History (Updated in Django 2.2)
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Unlike with ``bulk_create``, `queryset updates`_ perform an SQL update query on
 the queryset, and never return the actual updated objects (which would be
@@ -81,7 +92,7 @@ As the Django documentation says::
 
 .. _queryset updates: https://docs.djangoproject.com/en/2.2/ref/models/querysets/#update
 
-Note: Django 2.2 now allows ``bulk_update``. No ``pre_save`` or ``post_save`` signals are sent still. 
+Note: Django 2.2 now allows ``bulk_update``. No ``pre_save`` or ``post_save`` signals are sent still.
 
 Tracking Custom Users
 ---------------------

simple_history/manager.py

@@ -98,7 +98,14 @@ def _as_of_set(self, date):
                 continue
             yield last_change.instance
 
-    def bulk_history_create(self, objs, batch_size=None, update=False):
+    def bulk_history_create(
+        self,
+        objs,
+        batch_size=None,
+        update=False,
+        default_user=None,
+        default_change_reason="",
+    ):
         """
         Bulk create the history for the objects specified by objs.
         If called by bulk_update_with_history, use the update boolean and
@@ -113,8 +120,10 @@ def bulk_history_create(self, objs, batch_size=None, update=False):
         for instance in objs:
             row = self.model(
                 history_date=getattr(instance, "_history_date", timezone.now()),
-                history_user=getattr(instance, "_history_user", None),
-                history_change_reason=getattr(instance, "changeReason", ""),
+                history_user=getattr(instance, "_history_user", default_user),
+                history_change_reason=getattr(
+                    instance, "changeReason", default_change_reason
+                ),
                 history_type=history_type,
                 **{
                     field.attname: getattr(instance, field.attname)

simple_history/tests/tests/test_manager.py

@@ -126,6 +126,55 @@ def test_bulk_history_create_with_change_reason(self):
             )
         )
 
+    def test_bulk_history_create_with_default_user(self):
+        user = User.objects.create_user("tester", "tester@example.com")
+
+        Poll.history.bulk_history_create(self.data, default_user=user)
+
+        self.assertTrue(
+            all([history.history_user == user for history in Poll.history.all()])
+        )
+
+    def test_bulk_history_create_with_default_change_reason(self):
+        Poll.history.bulk_history_create(self.data, default_change_reason="test")
+
+        self.assertTrue(
+            all(
+                [
+                    history.history_change_reason == "test"
+                    for history in Poll.history.all()
+                ]
+            )
+        )
+
+    def test_bulk_history_create_history_user_overrides_default(self):
+        user1 = User.objects.create_user("tester1", "tester1@example.com")
+        user2 = User.objects.create_user("tester2", "tester2@example.com")
+
+        for data in self.data:
+            data._history_user = user1
+
+        Poll.history.bulk_history_create(self.data, default_user=user2)
+
+        self.assertTrue(
+            all([history.history_user == user1 for history in Poll.history.all()])
+        )
+
+    def test_bulk_history_create_change_reason_overrides_default(self):
+        for data in self.data:
+            data.changeReason = "my_reason"
+
+        Poll.history.bulk_history_create(self.data, default_change_reason="test")
+
+        self.assertTrue(
+            all(
+                [
+                    history.history_change_reason == "my_reason"
+                    for history in Poll.history.all()
+                ]
+            )
+        )
+
     def test_bulk_history_create_on_objs_without_ids(self):
         self.data = [
             Poll(question="Question 1", pub_date=datetime.now()),

simple_history/tests/tests/test_utils.py

@@ -1,3 +1,7 @@
+from unittest import skipIf
+
+import django
+from django.contrib.auth import get_user_model
 from django.db import IntegrityError
 from django.test import TestCase, TransactionTestCase
 from django.utils import timezone
@@ -11,7 +15,13 @@
     PollWithExcludeFields,
     Street,
 )
-from simple_history.utils import bulk_create_with_history, update_change_reason
+from simple_history.utils import (
+    bulk_create_with_history,
+    update_change_reason,
+    bulk_update_with_history,
+)
+
+User = get_user_model()
 
 
 class BulkCreateWithHistoryTestCase(TestCase):
@@ -37,6 +47,29 @@ def test_bulk_create_history(self):
         self.assertEqual(Poll.objects.count(), 5)
         self.assertEqual(Poll.history.count(), 5)
 
+    def test_bulk_create_history_with_default_user(self):
+        user = User.objects.create_user("tester", "tester@example.com")
+
+        bulk_create_with_history(self.data, Poll, default_user=user)
+
+        self.assertTrue(
+            all([history.history_user == user for history in Poll.history.all()])
+        )
+
+    def test_bulk_create_history_with_default_change_reason(self):
+        bulk_create_with_history(
+            self.data, Poll, default_change_reason="my change reason"
+        )
+
+        self.assertTrue(
+            all(
+                [
+                    history.history_change_reason == "my change reason"
+                    for history in Poll.history.all()
+                ]
+            )
+        )
+
     def test_bulk_create_history_num_queries_is_two(self):
         with self.assertNumQueries(2):
             bulk_create_with_history(self.data, Poll)
@@ -142,9 +175,95 @@ def test_bulk_create_no_ids_return(self, hist_manager_mock):
         result = bulk_create_with_history(objects, model)
         self.assertEqual(result, objects)
         hist_manager_mock().bulk_history_create.assert_called_with(
-            objects, batch_size=None
+            objects, batch_size=None, default_user=None, default_change_reason=None
+        )
+
+
+@skipIf(django.VERSION < (2, 2,), reason="bulk_update does not exist before 2.2")
+class BulkUpdateWithHistoryTestCase(TestCase):
+    def setUp(self):
+        self.data = [
+            Poll(id=1, question="Question 1", pub_date=timezone.now()),
+            Poll(id=2, question="Question 2", pub_date=timezone.now()),
+            Poll(id=3, question="Question 3", pub_date=timezone.now()),
+            Poll(id=4, question="Question 4", pub_date=timezone.now()),
+            Poll(id=5, question="Question 5", pub_date=timezone.now()),
+        ]
+        bulk_create_with_history(self.data, Poll)
+
+        self.data[3].question = "Updated question"
+
+    def test_bulk_update_history(self):
+        bulk_update_with_history(
+            self.data, Poll, fields=["question"],
+        )
+
+        self.assertEqual(Poll.objects.count(), 5)
+        self.assertEqual(Poll.objects.get(id=4).question, "Updated question")
+        self.assertEqual(Poll.history.count(), 10)
+        self.assertEqual(Poll.history.filter(history_type="~").count(), 5)
+
+    def test_bulk_update_history_with_default_user(self):
+        user = User.objects.create_user("tester", "tester@example.com")
+
+        bulk_update_with_history(
+            self.data, Poll, fields=["question"], default_user=user
+        )
+
+        self.assertTrue(
+            all(
+                [
+                    history.history_user == user
+                    for history in Poll.history.filter(history_type="~")
+                ]
+            )
+        )
+
+    def test_bulk_update_history_with_default_change_reason(self):
+        bulk_update_with_history(
+            self.data,
+            Poll,
+            fields=["question"],
+            default_change_reason="my change reason",
+        )
+
+        self.assertTrue(
+            all(
+                [
+                    history.history_change_reason == "my change reason"
+                    for history in Poll.history.filter(history_type="~")
+                ]
+            )
         )
 
+    def test_bulk_update_history_num_queries_is_two(self):
+        with self.assertNumQueries(2):
+            bulk_update_with_history(
+                self.data, Poll, fields=["question"],
+            )
+
+    def test_bulk_update_history_on_model_without_history_raises_error(self):
+        self.data = [
+            Place(id=1, name="Place 1"),
+            Place(id=2, name="Place 2"),
+            Place(id=3, name="Place 3"),
+        ]
+        Place.objects.bulk_create(self.data)
+        self.data[0].name = "test"
+
+        with self.assertRaises(NotHistoricalModelError):
+            bulk_update_with_history(self.data, Place, fields=["name"])
+
+    def test_num_queries_when_batch_size_is_less_than_total(self):
+        with self.assertNumQueries(6):
+            bulk_update_with_history(self.data, Poll, fields=["question"], batch_size=2)
+
+    def test_bulk_update_history_with_batch_size(self):
+        bulk_update_with_history(self.data, Poll, fields=["question"], batch_size=2)
+
+        self.assertEqual(Poll.objects.count(), 5)
+        self.assertEqual(Poll.history.filter(history_type="~").count(), 5)
+
 
 class UpdateChangeReasonTestCase(TestCase):
     def test_update_change_reason_with_excluded_fields(self):

simple_history/utils.py

@@ -42,7 +42,9 @@ def get_history_model_for_model(model):
     return get_history_manager_for_model(model).model
 
 
-def bulk_create_with_history(objs, model, batch_size=None):
+def bulk_create_with_history(
+    objs, model, batch_size=None, default_user=None, default_change_reason=None
+):
     """
     Bulk create the objects specified by objs while also bulk creating
     their history (all in one transaction).
@@ -52,6 +54,10 @@ def bulk_create_with_history(objs, model, batch_size=None):
     :param objs: List of objs (not yet saved to the db) of type model
     :param model: Model class that should be created
     :param batch_size: Number of objects that should be created in each batch
+    :param default_user: Optional user to specify as the history_user in each historical
+        record
+    :param default_change_reason: Optional change reason to specify as the change_reason
+        in each historical record
     :return: List of objs with IDs
     """
 
@@ -61,7 +67,12 @@ def bulk_create_with_history(objs, model, batch_size=None):
         objs_with_id = model.objects.bulk_create(objs, batch_size=batch_size)
         if objs_with_id and objs_with_id[0].pk:
             second_transaction_required = False
-            history_manager.bulk_history_create(objs_with_id, batch_size=batch_size)
+            history_manager.bulk_history_create(
+                objs_with_id,
+                batch_size=batch_size,
+                default_user=default_user,
+                default_change_reason=default_change_reason,
+            )
     if second_transaction_required:
         obj_list = []
         with transaction.atomic(savepoint=False):
@@ -70,19 +81,30 @@ def bulk_create_with_history(objs, model, batch_size=None):
                     filter(lambda x: x[1] is not None, model_to_dict(obj).items())
                 )
                 obj_list += model.objects.filter(**attributes)
-            history_manager.bulk_history_create(obj_list, batch_size=batch_size)
+            history_manager.bulk_history_create(
+                obj_list,
+                batch_size=batch_size,
+                default_user=default_user,
+                default_change_reason=default_change_reason,
+            )
         objs_with_id = obj_list
     return objs_with_id
 
 
-def bulk_update_with_history(objs, model, fields, batch_size=None):
+def bulk_update_with_history(
+    objs, model, fields, batch_size=None, default_user=None, default_change_reason=None,
+):
     """
     Bulk update the objects specified by objs while also bulk creating
     their history (all in one transaction).
     :param objs: List of objs of type model to be updated
     :param model: Model class that should be updated
     :param fields: The fields that are updated
     :param batch_size: Number of objects that should be updated in each batch
+    :param default_user: Optional user to specify as the history_user in each historical
+        record
+    :param default_change_reason: Optional change reason to specify as the change_reason
+        in each historical record
     """
     if django.VERSION < (2, 2,):
         raise NotImplementedError(
@@ -91,5 +113,13 @@ def bulk_update_with_history(objs, model, fields, batch_size=None):
         )
     history_manager = get_history_manager_for_model(model)
     with transaction.atomic(savepoint=False):
-        model.objects.bulk_update(objs, fields, batch_size=batch_size)
-        history_manager.bulk_history_create(objs, batch_size=batch_size, update=True)
+        model.objects.bulk_update(
+            objs, fields, batch_size=batch_size,
+        )
+        history_manager.bulk_history_create(
+            objs,
+            batch_size=batch_size,
+            update=True,
+            default_user=default_user,
+            default_change_reason=default_change_reason,
+        )