perf(importers): batch endpoint creation and status updates during import/reimport (#14489)

* perf(importers): batch endpoint creation and status updates during import/reimport

Replace per-finding endpoint_get_or_create() calls with a stateful EndpointManager
that accumulates endpoints and statuses across findings and flushes them in bulk at
batch boundaries. Reduces ~6200 DB queries to ~3-5 for a 1000-finding scan with 5
endpoints per finding (200 unique).

- EndpointManager now takes a `product` param and holds internal accumulators
- `record_endpoint()` deduplicates by normalized key within a batch
- `record_status_for_create()` / `record_statuses_to_mitigate()` / `record_statuses_to_reactivate()` accumulate operations
- `persist()` flushes all pending creates and bulk_updates in one shot
- `update_endpoint_status()` accumulates mitigate/reactivate lists instead of dispatching per-finding Celery tasks
- Removed old `chunk_endpoints_and_*`, `add_endpoints_to_unsaved_finding`, `mitigate_endpoint_status`, `reactivate_endpoint_status` methods
- Added unit tests for `_make_endpoint_unique_tuple` normalization
- Updated performance test fixture to match new stateful manager interface

* fix endpoint manager initialization

* fix(importers): restore tag inheritance and endpoint_manager init for direct callers

- bulk_create bypasses Django post_save signals, so manually call inherit_instance_tags()
  for each newly created Endpoint to preserve product tag inheritance behavior
- Initialize endpoint_manager in create_test() so callers that invoke create_test() +
  process_findings() directly (without going through process_scan()) don't hit a NoneType error

* refactor(importers): explicit test param for endpoint manager, rename and improve get_or_create_endpoints

- Add `test` parameter to `_create_endpoint_manager()` so callers pass
  the test explicitly instead of relying on `self.test` being set
- Raise `ValueError` with a clear message when `test is None`
- Initialize endpoint_manager in `DefaultImporter.process_scan()` after
  `parse_findings()` (which calls `create_test()` for fresh imports),
  covering both the sync path and the async Celery task path
- Rename `_fetch_and_create_endpoints()` → `get_or_create_endpoints()`
- Change return type to `tuple[dict, list]`: returns `(endpoints_by_key, created)`
  so callers know exactly which endpoints were newly inserted
- Rename local `key_to_endpoint` → `endpoints_by_key` for clarity

* ruff

* fix create endpoint manager

* fix counts

* refactor(importers): initialize EndpointManager eagerly in __init__

Instead of lazily initializing endpoint_manager in process_scan() or
create_test(), initialize it immediately in __init__ using the product
from the required engagement/test parameter. This eliminates the NoneType
AttributeError when callers invoke create_test() + process_findings()
directly without going through process_scan().

- DefaultImporter: uses self.engagement.product (engagement is required)
- DefaultReImporter: uses self.test.engagement.product (test is required)
- Remove _create_endpoint_manager() factory methods and lazy init sites
- Remove self.endpoint_manager = None from BaseImporter.__init__

* add test case

* ruff

* remove mock

* fix: replace status_finding_non_special prefetch with Python filtering

Drop the to_attr Prefetch for status_finding_non_special and instead
prefetch all endpoint statuses, filtering non-special ones in Python
via EndpointManager.get_non_special_endpoint_statuses().

This avoids AttributeError when a finding created during the same
reimport batch (via add_new_finding_to_candidates) is matched by a
subsequent finding — such findings were never loaded through the
prefetch queryset and lacked the to_attr attribute.

See: #14569

Author: valentijnscholten

dojo/finding/deduplication.py

@@ -336,17 +336,18 @@ def build_candidate_scope_queryset(test, mode="deduplication", service=None):
         # Base prefetches for both modes
         prefetch_list = ["endpoints", "vulnerability_id_set", "found_by"]
 
-        # Additional prefetches for reimport mode: fetch only non-special endpoint statuses with their
-        # endpoint joined in, so endpoint_manager can read status_finding_non_special directly without
-        # any extra DB queries
+        # Prefetch all endpoint statuses with their endpoint for reimport mode.
+        # The non-special filtering (excluding false_positive, out_of_scope, risk_accepted)
+        # is done in Python by EndpointManager.get_non_special_endpoint_statuses().
+        # We avoid using to_attr here because findings created during the same reimport
+        # batch (via add_new_finding_to_candidates) are never loaded through this queryset
+        # and would lack the to_attr, causing an AttributeError.
+        # See: https://github.com/DefectDojo/django-DefectDojo/pull/14569
         if mode == "reimport":
             prefetch_list.append(
                 Prefetch(
                     "status_finding",
-                    queryset=Endpoint_Status.objects.exclude(
-                        Q(false_positive=True) | Q(out_of_scope=True) | Q(risk_accepted=True),
-                    ).select_related("endpoint"),
-                    to_attr="status_finding_non_special",
+                    queryset=Endpoint_Status.objects.select_related("endpoint"),
                 ),
             )
 

dojo/importers/base_importer.py

@@ -14,7 +14,6 @@
 import dojo.finding.helper as finding_helper
 import dojo.risk_acceptance.helper as ra_helper
 from dojo.celery_dispatch import dojo_dispatch_task
-from dojo.importers.endpoint_manager import EndpointManager
 from dojo.importers.location_manager import LocationManager, UnsavedLocation
 from dojo.importers.options import ImporterOptions
 from dojo.jira_link.helper import is_keep_in_sync_with_jira
@@ -83,9 +82,6 @@ def __init__(
         ImporterOptions.__init__(self, *args, **kwargs)
         if settings.V3_FEATURE_LOCATIONS:
             self.location_manager = LocationManager()
-        else:
-            # TODO: Delete this after the move to Locations
-            self.endpoint_manager = EndpointManager()
 
     def check_child_implementation_exception(self):
         """
@@ -825,12 +821,17 @@ def process_endpoints(
             msg = "BaseImporter#process_endpoints() method is deprecated when V3_FEATURE_LOCATIONS is enabled"
             raise NotImplementedError(msg)
 
-        # Save the unsaved endpoints
-        self.endpoint_manager.chunk_endpoints_and_disperse(finding, finding.unsaved_endpoints)
-        # Check for any that were added in the form
+        # Clean and record unsaved endpoints from the report
+        self.endpoint_manager.clean_unsaved_endpoints(finding.unsaved_endpoints)
+        for endpoint in finding.unsaved_endpoints:
+            key = self.endpoint_manager.record_endpoint(endpoint)
+            self.endpoint_manager.record_status_for_create(finding, key)
+        # Record any endpoints added from the form
         if len(endpoints_to_add) > 0:
             logger.debug("endpoints_to_add: %s", endpoints_to_add)
-            self.endpoint_manager.chunk_endpoints_and_disperse(finding, endpoints_to_add)
+            for endpoint in endpoints_to_add:
+                key = self.endpoint_manager.record_endpoint(endpoint)
+                self.endpoint_manager.record_status_for_create(finding, key)
 
     def sanitize_vulnerability_ids(self, finding) -> None:
         """Remove undisired vulnerability id values"""
@@ -934,14 +935,8 @@ def mitigate_finding(
             )
         else:
             # TODO: Delete this after the move to Locations
-            # Mitigate the endpoint statuses
-            dojo_dispatch_task(
-                EndpointManager.mitigate_endpoint_status,
-                finding.status_finding.all(),
-                self.user,
-                kwuser=self.user,
-                sync=True,
-            )
+            # Accumulate endpoint statuses for bulk mitigate in persist()
+            self.endpoint_manager.record_statuses_to_mitigate(finding.status_finding.all())
         # to avoid pushing a finding group multiple times, we push those outside of the loop
         if finding_groups_enabled and finding.finding_group:
             # don't try to dedupe findings that we are closing

dojo/importers/default_importer.py

@@ -9,6 +9,7 @@
 from dojo.celery_dispatch import dojo_dispatch_task
 from dojo.finding import helper as finding_helper
 from dojo.importers.base_importer import BaseImporter, Parser
+from dojo.importers.endpoint_manager import EndpointManager
 from dojo.importers.options import ImporterOptions
 from dojo.jira_link.helper import is_keep_in_sync_with_jira
 from dojo.models import (
@@ -56,6 +57,8 @@ def __init__(self, *args, **kwargs):
             import_type=Test_Import.IMPORT_TYPE,
             **kwargs,
         )
+        if not settings.V3_FEATURE_LOCATIONS:
+            self.endpoint_manager = EndpointManager(self.engagement.product)
 
     def create_test(
         self,
@@ -109,6 +112,8 @@ def process_scan(
         parser = self.get_parser()
         # Get the findings from the parser based on what methods the parser supplies
         # This could either mean traditional file parsing, or API pull parsing
+        # Note: for fresh imports, parse_findings() calls create_test() internally,
+        # so self.test is guaranteed to be set after this call.
         parsed_findings = self.parse_findings(scan, parser) or []
         new_findings = self.process_findings(parsed_findings, **kwargs)
         # Close any old findings in the processed list if the the user specified for that
@@ -259,8 +264,10 @@ def process_findings(
             logger.debug("process_findings: computed push_to_jira=%s", push_to_jira)
             batch_finding_ids.append(finding.id)
 
-            # If batch is full or we're at the end, dispatch one batched task
+            # If batch is full or we're at the end, persist endpoints and dispatch
             if len(batch_finding_ids) >= batch_max_size or is_final_finding:
+                if not settings.V3_FEATURE_LOCATIONS:
+                    self.endpoint_manager.persist(user=self.user)
                 finding_ids_batch = list(batch_finding_ids)
                 batch_finding_ids.clear()
                 logger.debug("process_findings: dispatching batch with push_to_jira=%s (batch_size=%d, is_final=%s)",
@@ -378,6 +385,9 @@ def close_old_findings(
                 finding_groups_enabled=self.findings_groups_enabled,
                 product_grading_option=False,
             )
+        # Persist any accumulated endpoint status mitigations
+        if not settings.V3_FEATURE_LOCATIONS:
+            self.endpoint_manager.persist(user=self.user)
         # push finding groups to jira since we only only want to push whole groups
         # We dont check if the finding jira sync is applicable quite yet until we can get in the loop
         # but this is a way to at least make it that far

dojo/importers/default_reimporter.py

@@ -14,6 +14,7 @@
     find_candidates_for_reimport_legacy,
 )
 from dojo.importers.base_importer import BaseImporter, Parser
+from dojo.importers.endpoint_manager import EndpointManager
 from dojo.importers.options import ImporterOptions
 from dojo.jira_link.helper import is_keep_in_sync_with_jira
 from dojo.location.status import FindingLocationStatus
@@ -76,6 +77,8 @@ def __init__(self, *args, **kwargs):
             import_type=Test_Import.REIMPORT_TYPE,
             **kwargs,
         )
+        if not settings.V3_FEATURE_LOCATIONS:
+            self.endpoint_manager = EndpointManager(self.test.engagement.product)
 
     def process_scan(
         self,
@@ -430,6 +433,8 @@ def process_findings(
                     # - Deduplication batches: optimize bulk operations (larger batches = fewer queries)
                     # They don't need to be aligned since they optimize different operations.
                     if len(batch_finding_ids) >= dedupe_batch_max_size or is_final:
+                        if not settings.V3_FEATURE_LOCATIONS:
+                            self.endpoint_manager.persist(user=self.user)
                         finding_ids_batch = list(batch_finding_ids)
                         batch_finding_ids.clear()
                         dojo_dispatch_task(
@@ -497,6 +502,9 @@ def close_old_findings(
                     product_grading_option=False,
                 )
                 mitigated_findings.append(finding)
+        # Persist any accumulated endpoint status mitigations
+        if not settings.V3_FEATURE_LOCATIONS:
+            self.endpoint_manager.persist(user=self.user)
         # push finding groups to jira since we only only want to push whole groups
         # We dont check if the finding jira sync is applicable quite yet until we can get in the loop
         # but this is a way to at least make it that far
@@ -763,9 +771,10 @@ def process_matched_mitigated_finding(
             self.location_manager.chunk_locations_and_reactivate(mitigated_locations)
         else:
             # TODO: Delete this after the move to Locations
-            # Reactivate mitigated endpoints that are not false positives, out of scope, or risk accepted
-            # status_finding_non_special is prefetched by build_candidate_scope_queryset
-            self.endpoint_manager.chunk_endpoints_and_reactivate(existing_finding.status_finding_non_special)
+            # Accumulate endpoint statuses for bulk reactivation in persist()
+            self.endpoint_manager.record_statuses_to_reactivate(
+                self.endpoint_manager.get_non_special_endpoint_statuses(existing_finding),
+            )
         existing_finding.notes.add(note)
         self.reactivated_items.append(existing_finding)
         # The new finding is active while the existing on is mitigated. The existing finding needs to
@@ -932,9 +941,13 @@ def finding_post_processing(
                 self.location_manager.chunk_locations_and_disperse(finding, self.endpoints_to_add)
         else:
             # TODO: Delete this after the move to Locations
-            self.endpoint_manager.chunk_endpoints_and_disperse(finding, finding_from_report.unsaved_endpoints)
+            for endpoint in finding_from_report.unsaved_endpoints:
+                key = self.endpoint_manager.record_endpoint(endpoint)
+                self.endpoint_manager.record_status_for_create(finding, key)
             if len(self.endpoints_to_add) > 0:
-                self.endpoint_manager.chunk_endpoints_and_disperse(finding, self.endpoints_to_add)
+                for endpoint in self.endpoints_to_add:
+                    key = self.endpoint_manager.record_endpoint(endpoint)
+                    self.endpoint_manager.record_status_for_create(finding, key)
         # Parsers shouldn't use the tags field, and use unsaved_tags instead.
         # Merge any tags set by parser into unsaved_tags
         tags_from_parser = finding_from_report.tags if isinstance(finding_from_report.tags, list) else []