Add cascading for model delete

bulk_tracker/models.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
-from django.db import models
+from django.db import models, router
 from model_utils import FieldTracker
 
+from bulk_tracker.collector import BulkTrackerCollector
 from bulk_tracker.helper_objects import TrackingInfo
 from bulk_tracker.managers import BulkTrackerManager
 from bulk_tracker.signals import (
@@ -40,6 +41,18 @@ def save(self, tracking_info_: TrackingInfo | None = None, **kwargs):
                 f"Model {self.__class__} doesn't have tracker, please add `tracker = FieldTracker()` to your model"
             )
 
-    def delete(self, *args, tracking_info_: TrackingInfo | None = None, **kwargs):
+    def delete(self, using=None, keep_parents=False, tracking_info_: TrackingInfo | None = None):
+        if self.pk is None:
+            raise ValueError(
+                "%s object can't be deleted because its %s attribute is set "
+                "to None." % (self._meta.object_name, self._meta.pk.attname)
+            )
+        using = using or router.db_for_write(self.__class__, instance=self)
+        try:
+            collector = BulkTrackerCollector(using=using, origin=self)
+        except TypeError:
+            collector = BulkTrackerCollector(using=using)
+        collector.collect([self], keep_parents=keep_parents)
+        ret = collector.delete(tracking_info_=tracking_info_)
         send_post_delete_signal([self], model=self.__class__, tracking_info_=tracking_info_)
-        return super().delete(*args, **kwargs)
+        return ret

tests/test_delete_signal.py

@@ -125,7 +125,7 @@ def test_should_use_robust_send_if_is_robust_is_true_in_tracking_info(self, mock
         mocked_signal.assert_not_called()
         mocked_signal_robust.assert_called_once()
 
-    def test_should_send_post_delete_signal_for_foreign_keys_with_cascade_if_fast_delete_is_used(self):
+    def test_queryset_delete_should_send_post_delete_signal_for_foreign_keys_with_cascade_if_fast_delete_is_used(self):
         # Arrange
         signal_called_with = {}
 
@@ -153,7 +153,9 @@ def post_delete_receiver(
         self.assertEqual(datetime.strptime("1998-01-08", "%Y-%m-%d").date(), modified_objects[0].instance.publish_date)
         self.assertEqual(self.author_john.id, modified_objects[0].instance.author_id)
 
-    def test_should_send_post_delete_signal_for_foreign_keys_with_cascade_if_fast_delete_is_not_used(self):
+    def test_queryset_delete_should_send_post_delete_signal_for_foreign_keys_with_cascade_if_fast_delete_is_not_used(
+        self,
+    ):
         # Arrange
         signal_called_with = {}
 
@@ -184,3 +186,73 @@ def post_delete_receiver(
         self.assertEqual("Sound of Winter", modified_objects[0].instance.title)
         self.assertEqual(datetime.strptime("1998-01-08", "%Y-%m-%d").date(), modified_objects[0].instance.publish_date)
         self.assertEqual(self.author_john.id, modified_objects[0].instance.author_id)
+
+    def test_model_delete_should_send_post_delete_signal_for_foreign_keys_with_cascade_if_fast_delete_is_used(self):
+        # Arrange
+        signal_called_with = {}
+
+        def post_delete_receiver(
+            sender,
+            objects: list[ModifiedObject[Post]],
+            tracking_info_: TrackingInfo | None = None,
+            **kwargs,
+        ):
+            signal_called_with.setdefault("times_called", 0)
+            signal_called_with["times_called"] += 1
+            signal_called_with["sender"] = sender
+            signal_called_with["objects"] = objects
+            signal_called_with["tracking_info_"] = tracking_info_
+
+        post_delete_signal.connect(post_delete_receiver, sender=Post)
+        post = Post.objects.create(title="Sound of Winter", publish_date="1998-01-08", author=self.author_john)
+        author_id = self.author_john.id
+
+        # Act
+        self.author_john.delete(tracking_info_=TrackingInfo(comment="This is a comment"))
+
+        # Assert
+        modified_objects: list[ModifiedObject[Post]] = signal_called_with["objects"]
+        self.assertEqual(signal_called_with["sender"], Post)
+        self.assertEqual(post.id, modified_objects[0].instance.id)
+        self.assertEqual("Sound of Winter", modified_objects[0].instance.title)
+        self.assertEqual(datetime.strptime("1998-01-08", "%Y-%m-%d").date(), modified_objects[0].instance.publish_date)
+        self.assertEqual(author_id, modified_objects[0].instance.author_id)
+        self.assertEqual("This is a comment", signal_called_with["tracking_info_"].comment)
+        self.assertEqual(1, signal_called_with["times_called"])
+
+    def test_model_delete_should_send_post_delete_signal_for_foreign_keys_with_cascade_if_fast_delete_is_not_used(self):
+        # Arrange
+        signal_called_with = {}
+
+        def pre_delete_receiver(**kwargs):
+            pass
+
+        def post_delete_receiver(
+            sender,
+            objects: list[ModifiedObject[Post]],
+            tracking_info_: TrackingInfo | None = None,
+            **kwargs,
+        ):
+            signal_called_with.setdefault("times_called", 0)
+            signal_called_with["times_called"] += 1
+            signal_called_with["sender"] = sender
+            signal_called_with["objects"] = objects
+            signal_called_with["tracking_info_"] = tracking_info_
+
+        pre_delete.connect(pre_delete_receiver, sender=Post)  # disable fast delete in `Collector`
+        post_delete_signal.connect(post_delete_receiver, sender=Post)
+        post = Post.objects.create(title="Sound of Winter", publish_date="1998-01-08", author=self.author_john)
+        author_id = self.author_john.id
+
+        # Act
+        self.author_john.delete(tracking_info_=TrackingInfo(comment="This is a comment"))
+
+        # Assert
+        modified_objects: list[ModifiedObject[Post]] = signal_called_with["objects"]
+        self.assertEqual(signal_called_with["sender"], Post)
+        self.assertEqual(post.id, modified_objects[0].instance.id)
+        self.assertEqual("Sound of Winter", modified_objects[0].instance.title)
+        self.assertEqual(datetime.strptime("1998-01-08", "%Y-%m-%d").date(), modified_objects[0].instance.publish_date)
+        self.assertEqual(author_id, modified_objects[0].instance.author_id)
+        self.assertEqual("This is a comment", signal_called_with["tracking_info_"].comment)
+        self.assertEqual(1, signal_called_with["times_called"])