Add m2m signals and specific methods

Author: legau

docs/signals.rst

@@ -22,6 +22,24 @@ saving a historical record. Arguments passed to the signals include the followin
     using
         The database alias being used
 
+For Many To Many signals you've got the following :
+
+.. glossary::
+    instance
+        The source model instance being saved
+
+    history_instance
+        The corresponding history record
+
+    rows (for pre_create)
+        The elements to be bulk inserted into the m2m table
+
+    created_rows (for post_create)
+        The created elements into the m2m table
+
+    field
+        The recorded field object
+
 To connect the signals to your callbacks, you can use the ``@receiver`` decorator:
 
 .. code-block:: python
@@ -30,6 +48,8 @@ To connect the signals to your callbacks, you can use the ``@receiver`` decorato
     from simple_history.signals import (
         pre_create_historical_record,
         post_create_historical_record
+        pre_create_historical_m2m_records,
+        post_create_historical_m2m_records,
     )
 
     @receiver(pre_create_historical_record)
@@ -39,3 +59,11 @@ To connect the signals to your callbacks, you can use the ``@receiver`` decorato
     @receiver(post_create_historical_record)
     def post_create_historical_record_callback(sender, **kwargs):
         print("Sent after saving historical record")
+
+    @receiver(pre_create_historical_m2m_records)
+    def pre_create_historical_m2m_records_callback(sender, **kwargs):
+        print("Sent before saving many to many field on historical record")
+
+    @receiver(post_create_historical_m2m_records)
+    def post_create_historical_m2m_records_callback(sender, **kwargs):
+        print("Sent after saving many to many field on historical record")

simple_history/models.py

@@ -32,7 +32,12 @@
 
 from . import exceptions
 from .manager import SIMPLE_HISTORY_REVERSE_ATTR_NAME, HistoryDescriptor
-from .signals import post_create_historical_record, pre_create_historical_record
+from .signals import (
+    post_create_historical_m2m_records,
+    post_create_historical_record,
+    pre_create_historical_m2m_records,
+    pre_create_historical_record,
+)
 from .utils import get_change_reason_from_object
 
 try:
@@ -95,6 +100,7 @@ def __init__(
         excluded_field_kwargs=None,
         m2m_fields=(),
         m2m_fields_model_field_name="_history_m2m_fields",
+        m2m_bases=(models.Model,),
     ):
         self.user_set_verbose_name = verbose_name
         self.user_set_verbose_name_plural = verbose_name_plural
@@ -134,6 +140,12 @@ def __init__(
             self.bases = (HistoricalChanges,) + tuple(bases)
         except TypeError:
             raise TypeError("The `bases` option must be a list or a tuple.")
+        try:
+            if isinstance(m2m_bases, str):
+                raise TypeError
+            self.m2m_bases = (HistoricalChanges,) + tuple(m2m_bases)
+        except TypeError:
+            raise TypeError("The `m2m_bases` option must be a list or a tuple.")
 
     def contribute_to_class(self, cls, name):
         self.manager_name = name
@@ -239,35 +251,18 @@ def get_history_model_name(self, model):
         )
 
     def create_history_m2m_model(self, model, through_model):
-        attrs = {
-            "__module__": model.__module__,
-            "__str__": lambda self: "{} as of {}".format(
-                self._meta.verbose_name, self.history.history_date
-            ),
-        }
-
-        # Get the primary key to the history model this model will look up to
-        attrs["m2m_history_id"] = self._get_history_id_field()
-        attrs["history"] = models.ForeignKey(
-            model,
-            db_constraint=False,
-            on_delete=models.DO_NOTHING,
-        )
-        attrs["instance_type"] = through_model
+        attrs = {}
 
         fields = self.copy_fields(through_model)
         attrs.update(fields)
+        attrs.update(self.get_extra_fields_m2m(model, through_model, fields))
 
         name = self.get_history_model_name(through_model)
         registered_models[through_model._meta.db_table] = through_model
-        meta_fields = {"verbose_name": name}
-
-        if self.app:
-            meta_fields["app_label"] = self.app
 
-        attrs.update(Meta=type(str("Meta"), (), meta_fields))
+        attrs.update(Meta=type("Meta", (), self.get_meta_options_m2m(through_model)))
 
-        m2m_history_model = type(str(name), (models.Model,), attrs)
+        m2m_history_model = type(str(name), self.m2m_bases, attrs)
 
         return m2m_history_model
 
@@ -458,6 +453,25 @@ def _get_history_related_field(self, model):
         else:
             return {}
 
+    def get_extra_fields_m2m(self, model, through_model, fields):
+        """Return dict of extra fields added to the m2m historical record model"""
+
+        extra_fields = {
+            "__module__": model.__module__,
+            "__str__": lambda self: "{} as of {}".format(
+                self._meta.verbose_name, self.history.history_date
+            ),
+            "history": models.ForeignKey(
+                model,
+                db_constraint=False,
+                on_delete=models.DO_NOTHING,
+            ),
+            "instance_type": through_model,
+            "m2m_history_id": self._get_history_id_field(),
+        }
+
+        return extra_fields
+
     def get_extra_fields(self, model, fields):
         """Return dict of extra fields added to the historical record model"""
 
@@ -570,6 +584,20 @@ def _date_indexing(self):
             )
         return result
 
+    def get_meta_options_m2m(self, through_model):
+        """
+        Returns a dictionary of fields that will be added to
+        the Meta inner class of the m2m historical record model.
+        """
+        name = self.get_history_model_name(through_model)
+
+        meta_fields = {"verbose_name": name}
+
+        if self.app:
+            meta_fields["app_label"] = self.app
+
+        return meta_fields
+
     def get_meta_options(self, model):
         """
         Returns a dictionary of fields that will be added to
@@ -650,7 +678,21 @@ def create_historical_record_m2ms(self, history_instance, instance):
                     )
                 insert_rows.append(m2m_history_model(**insert_row))
 
-            m2m_history_model.objects.bulk_create(insert_rows)
+            pre_create_historical_m2m_records.send(
+                sender=m2m_history_model,
+                rows=insert_rows,
+                history_instance=history_instance,
+                instance=instance,
+                field=field,
+            )
+            created_rows = m2m_history_model.objects.bulk_create(insert_rows)
+            post_create_historical_m2m_records.send(
+                sender=m2m_history_model,
+                created_rows=created_rows,
+                history_instance=history_instance,
+                instance=instance,
+                field=field,
+            )
 
     def create_historical_record(self, instance, history_type, using=None):
         using = using if self.use_base_model_db else None

simple_history/signals.py

@@ -7,3 +7,11 @@
 # Arguments: "instance",  "history_instance", "history_date",
 #            "history_user", "history_change_reason", "using"
 post_create_historical_record = django.dispatch.Signal()
+
+# Arguments: "sender",  "rows", "history_instance", "instance",
+#             "field"
+pre_create_historical_m2m_records = django.dispatch.Signal()
+
+# Arguments: "sender",  "created_rows", "history_instance",
+#            "instance", "field"
+post_create_historical_m2m_records = django.dispatch.Signal()

simple_history/tests/models.py

@@ -117,6 +117,27 @@ class PollWithManyToMany(models.Model):
     history = HistoricalRecords(m2m_fields=[places])
 
 
+class HistoricalRecordsWithExtraFieldM2M(HistoricalRecords):
+    def get_extra_fields_m2m(self, model, through_model, fields):
+        extra_fields = super().get_extra_fields_m2m(model, through_model, fields)
+
+        def get_class_name(self):
+            return self.__class__.__name__
+
+        extra_fields["get_class_name"] = get_class_name
+        return extra_fields
+
+
+class PollWithManyToManyWithIPAddress(models.Model):
+    question = models.CharField(max_length=200)
+    pub_date = models.DateTimeField("date published")
+    places = models.ManyToManyField("Place")
+
+    history = HistoricalRecordsWithExtraFieldM2M(
+        m2m_fields=[places], m2m_bases=[IPAddressHistoricalModel]
+    )
+
+
 class PollWithSeveralManyToMany(models.Model):
     question = models.CharField(max_length=200)
     pub_date = models.DateTimeField("date published")

simple_history/tests/tests/test_models.py

@@ -24,7 +24,10 @@
     is_historic,
     to_historic,
 )
-from simple_history.signals import pre_create_historical_record
+from simple_history.signals import (
+    pre_create_historical_m2m_records,
+    pre_create_historical_record,
+)
 from simple_history.tests.custom_user.models import CustomUser
 from simple_history.tests.tests.utils import (
     database_router_override_settings,
@@ -95,6 +98,7 @@
     PollWithExcludeFields,
     PollWithHistoricalIPAddress,
     PollWithManyToMany,
+    PollWithManyToManyWithIPAddress,
     PollWithNonEditableField,
     PollWithSeveralManyToMany,
     Province,
@@ -1487,6 +1491,11 @@ def add_static_history_ip_address(sender, **kwargs):
     history_instance.ip_address = "192.168.0.1"
 
 
+def add_static_history_ip_address_on_m2m(sender, rows, **kwargs):
+    for row in rows:
+        row.ip_address = "192.168.0.1"
+
+
 class ExtraFieldsStaticIPAddressTestCase(TestCase):
     def setUp(self):
         pre_create_historical_record.connect(
@@ -1790,6 +1799,55 @@ def test_separation(self):
         self.assertEqual(add.places.all().count(), 0)
 
 
+class ManyToManyWithSignalsTest(TestCase):
+    def setUp(self):
+        self.model = PollWithManyToManyWithIPAddress
+        # self.historical_through_model = self.model.history.
+        self.places = (
+            Place.objects.create(name="London"),
+            Place.objects.create(name="Paris"),
+        )
+        self.poll = self.model.objects.create(question="what's up?", pub_date=today)
+        pre_create_historical_m2m_records.connect(
+            add_static_history_ip_address_on_m2m,
+            dispatch_uid="add_static_history_ip_address_on_m2m",
+        )
+
+    def tearDown(self):
+        pre_create_historical_m2m_records.disconnect(
+            add_static_history_ip_address_on_m2m,
+            dispatch_uid="add_static_history_ip_address_on_m2m",
+        )
+
+    def test_ip_address_added(self):
+        self.poll.places.add(*self.places)
+
+        places = self.poll.history.first().places
+        self.assertEqual(2, places.count())
+        for place in places.all():
+            self.assertEqual("192.168.0.1", place.ip_address)
+
+    def test_extra_field(self):
+        self.poll.places.add(*self.places)
+        m2m_record = self.poll.history.first().places.first()
+        self.assertEqual(
+            m2m_record.get_class_name(),
+            "HistoricalPollWithManyToManyWithIPAddress_places",
+        )
+
+    def test_diff(self):
+        self.poll.places.clear()
+        self.poll.places.add(*self.places)
+
+        new = self.poll.history.first()
+        old = new.prev_record
+
+        delta = new.diff_against(old)
+
+        self.assertEqual("places", delta.changes[0].field)
+        self.assertEqual(2, len(delta.changes[0].new))
+
+
 class ManyToManyTest(TestCase):
     def setUp(self):
         self.model = PollWithManyToMany