Merge pull request #10 from mighty-justice/create-with-audit-trail

Add option to merge models with an audit trail

django_super_deduper/merge.py

@@ -1,5 +1,5 @@
 import logging
-from typing import List
+from typing import List, Tuple
 
 from django.core.exceptions import ValidationError
 from django.core.serializers import serialize
@@ -18,16 +18,16 @@ def __init__(self, primary_object: Model, keep_old=True, merge_field_values=True
         self.keep_old = keep_old
         self.merge_field_values = merge_field_values
         self.model_meta = ModelMeta(primary_object)
+        self.modified_related_objects = []  # type: List
 
     @classmethod
-    def create(
+    def _create(
         cls,
         primary_object: Model,
         alias_objects: List[Model],
         keep_old=True,
         merge_field_values=True,
-    ) -> Model:
-
+    ) -> 'MergedModelInstance':
         merged_model_instance = cls(primary_object, keep_old=keep_old, merge_field_values=merge_field_values)
 
         logger.debug(f'Primary object {merged_model_instance.model_meta.model_name}[pk={primary_object.pk}] '
@@ -38,7 +38,16 @@ def create(
         for alias_object in alias_objects:
             merged_model_instance.merge(alias_object)
 
-        return merged_model_instance.primary_object
+        return merged_model_instance
+
+    @classmethod
+    def create(cls, *args, **kwargs) -> Model:
+        return cls._create(*args, **kwargs).primary_object
+
+    @classmethod
+    def create_with_audit_trail(cls, *args, **kwargs) -> Tuple[Model, List[Model]]:
+        instance = cls._create(*args, **kwargs)
+        return instance.primary_object, instance.modified_related_objects
 
     def _handle_o2m_related_field(self, related_field: Field, alias_object: Model):
         reverse_o2m_accessor_name = related_field.get_accessor_name()
@@ -63,6 +72,7 @@ def _handle_o2m_related_field(self, related_field: Field, alias_object: Model):
                 else:
                     logger.debug(f'Deleting {obj._meta.model.__name__}[pk={obj.pk}]')
                     obj.delete()
+            self.modified_related_objects.append(obj)
 
     def _handle_m2m_related_field(self, related_field: Field, alias_object: Model):
         try:
@@ -78,6 +88,7 @@ def _handle_m2m_related_field(self, related_field: Field, alias_object: Model):
             logger.debug(f'Adding {obj._meta.model.__name__}[pk={obj.pk}] '
                          f'to {self.model_meta.model_name}[pk={self.primary_object.pk}].{m2m_accessor_name}')
             getattr(self.primary_object, m2m_accessor_name).add(obj)
+            self.modified_related_objects.append(obj)
 
     def _handle_o2o_related_field(self, related_field: Field, alias_object: Model):
         if not self.merge_field_values:
@@ -94,6 +105,7 @@ def _handle_o2o_related_field(self, related_field: Field, alias_object: Model):
             logger.debug(f'Setting {o2o_accessor_name} on {self.model_meta.model_name}[pk={self.primary_object.pk}] '
                          f'to {alias_o2o_object._meta.model.__name__}[pk={alias_o2o_object.pk}')
             setattr(self.primary_object, o2o_accessor_name, alias_o2o_object)
+            self.modified_related_objects.append(alias_o2o_object)
 
     def merge(self, alias_object: Model):
         primary_object = self.primary_object

tests/test_super_deduper.py

@@ -147,3 +147,40 @@ def test_prevent_self_merge(self):
 
         with pytest.raises(ValueError):
             MergedModelInstance.create(primary_object, [alias_object])
+
+    def test_o2o_merge_with_audit_trail(self):
+        primary_object = RestaurantFactory.create(place=None, serves_hot_dogs=True, serves_pizza=False)
+        alias_objects = RestaurantFactory.create_batch(3)
+        related_object = set([alias_objects[0].place])
+
+        _, audit_trail = MergedModelInstance.create_with_audit_trail(primary_object, alias_objects)
+
+        assert set(audit_trail) == related_object
+
+    def test_o2m_merge_with_audit_trail(self):
+        primary_object = NewsAgencyFactory.create()
+        alias_object = NewsAgencyFactory.create()
+        related_objects = set(ReporterFactory.create_batch(3, news_agency=alias_object))
+
+        _, audit_trail = MergedModelInstance.create_with_audit_trail(primary_object, [alias_object])
+
+        assert set(audit_trail) == related_objects
+
+    def test_m2m_merge_with_audit_trail(self):
+        primary_object = ArticleFactory.create(reporter=None)
+        related_object = ReporterFactory.create()
+        alias_object = ArticleFactory.create(number_of_publications=3, reporter=related_object)
+        related_objects = set(alias_object.publications.all())
+
+        _, audit_trail = MergedModelInstance.create_with_audit_trail(primary_object, [alias_object])
+
+        assert set(audit_trail) == related_objects
+
+    def test_reverse_m2m_merge_with_audit_trail(self):
+        primary_object = PublicationFactory.create()
+        alias_object = PublicationFactory.create(number_of_articles=3)
+        related_objects = set(alias_object.article_set.all())
+
+        _, audit_trail = MergedModelInstance.create_with_audit_trail(primary_object, [alias_object])
+
+        assert set(audit_trail) == related_objects