Add bulk delete findings extension hook and optional cascade context

- Split internal bulk delete implementation from entry point that can
  delegate via get_custom_method(BULK_DELETE_FINDINGS_METHOD).
- Allow callers to pass optional cascade_root alongside chunk_size for
  downstream use when a custom method is configured.

Author: valentijnscholten

dojo/finding/helper.py

@@ -764,7 +764,7 @@ def bulk_clear_finding_m2m(finding_qs):
         Notes.objects.filter(id__in=note_ids).delete()
 
 
-def bulk_delete_findings(finding_qs, chunk_size=1000):
+def _bulk_delete_findings_internal(finding_qs, chunk_size=1000):
     """
     Delete findings and all related objects efficiently. Including any related object in Dojo-Pro
 
@@ -799,6 +799,20 @@ def bulk_delete_findings(finding_qs, chunk_size=1000):
         )
 
 
+def bulk_delete_findings(finding_qs, chunk_size=1000, cascade_root=None):
+    """Entry point; may delegate to Pro via settings.BULK_DELETE_FINDINGS_METHOD.
+
+    cascade_root: optional dict describing the top-level object whose cascade triggered
+    this bulk delete (e.g. {"model": "dojo.engagement", "pk": 9}). Ignored by OSS
+    when no custom method is configured.
+    """
+    from dojo.utils import get_custom_method  # noqa: PLC0415 circular import
+
+    if fn := get_custom_method("BULK_DELETE_FINDINGS_METHOD"):
+        return fn(finding_qs, chunk_size=chunk_size, cascade_root=cascade_root)
+    return _bulk_delete_findings_internal(finding_qs, chunk_size=chunk_size)
+
+
 def fix_loop_duplicates(scope_qs=None):
     """Due to bugs in the past and even currently when under high parallel load, there can be transitive duplicates."""
     """ i.e. A -> B -> C. This can lead to problems when deleting findingns, performing deduplication, etc """

dojo/utils.py

@@ -2024,6 +2024,8 @@ def async_delete_task(obj, **kwargs):
     scope_field = FINDING_SCOPE_FILTERS.get(type(obj))
     if scope_field:
         finding_qs = Finding.objects.filter(**{scope_field: obj})
+        # cascade_root is some context we provide to the bulk_delete_findings function
+        cascade_root = {"model": obj._meta.label_lower, "pk": obj.pk}
 
         # Step 2: Prepare duplicate clusters (must happen before any deletion)
         # When CASCADE_DELETE=True, reconfigure_duplicate_cluster skips reconfiguration —
@@ -2042,10 +2044,10 @@ def async_delete_task(obj, **kwargs):
         outside_count = outside_dupes_qs.count()
         if outside_count:
             logger.info("ASYNC_DELETE: Deleting %d outside-scope duplicates first", outside_count)
-            bulk_delete_findings(outside_dupes_qs, chunk_size=chunk_size)
+            bulk_delete_findings(outside_dupes_qs, chunk_size=chunk_size, cascade_root=cascade_root)
 
         # Step 4: Delete the main scope findings
-        bulk_delete_findings(finding_qs, chunk_size=chunk_size)
+        bulk_delete_findings(finding_qs, chunk_size=chunk_size, cascade_root=cascade_root)
 
     # Step 5: Delete all remaining related objects (Tests, Engagements,
     # Endpoints, etc.) via SQL cascade. Findings are already gone, so