Coverage for netbox/models/deletion.py: 70%
58 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import logging
3from django.contrib.contenttypes.fields import GenericRelation
4from django.db import router
5from django.db.models.deletion import CASCADE, Collector
6from django.utils.translation import gettext as _
8logger = logging.getLogger("netbox.models.deletion")
11class CountOnly:
12 """
13 A stand-in for a list of dependent instances that reports a count without holding any
14 instances. Used on the delete-confirmation page for high-cardinality relations (e.g. a
15 JobsMixin object's jobs) which we deliberately do not materialize (see #22812). It is a
16 lenient, empty iterable: `len()` returns the true row count, but iterating yields nothing,
17 so it slots into the same `{model: <iterable>}` mapping as real instance lists and renders
18 as a non-expandable row.
19 """
20 # Template flag: distinguishes a count-only entry (no instances to list) from a real list,
21 # so the confirmation page can render it without an expand/collapse affordance.
22 count_only = True
24 def __init__(self, count):
25 self.count = count
27 def __len__(self):
28 return self.count
30 def __iter__(self):
31 return iter(())
34class ConfirmCollector(Collector):
35 """
36 A display-only Collector used to enumerate the objects that would be deleted along with a
37 given object, for rendering the delete confirmation page. It behaves like Django's stock
38 Collector (preserving the full FK cascade graph and its ProtectedError/RestrictedError
39 behavior) except that it does not descend into the `jobs` GenericRelation. A JobsMixin
40 object can accumulate thousands of Jobs, each carrying large data/log_entries payloads;
41 materializing them all just to render a confirmation page can exhaust memory (see #22812).
42 Instead, the related Jobs are counted and recorded in `generic_relation_counts`.
44 This is intentionally specific to Job, the only high-cardinality GenericRelation in the
45 data model; it is not a general count-out over every GenericRelation. If another relation
46 ever needs the same treatment, extend the check in collect() (and the matching write-path
47 batching in JobsMixin/ScriptModule.delete) rather than assuming this already handles it.
48 """
49 def __init__(self, *args, **kwargs):
50 super().__init__(*args, **kwargs)
51 self.generic_relation_counts = {}
53 def collect(self, objs, source=None, *args, **kwargs):
54 """
55 Override collect() to count the `jobs` GenericRelation rather than descend into it.
57 Django's Collector offers no per-relation skip hook, so we intercept the one call it
58 makes when cascading into a GenericRelation: collect(sub_objs, source=model, ...), where
59 `sub_objs` is a queryset of the related model. When that model is Job, we count the rows
60 instead of collecting (and thus instantiating) them, and forward every other call to the
61 stock implementation untouched. A directly-deleted Job (top-level call, source=None)
62 still collects normally.
63 """
64 from core.models import Job
66 if source is not None and getattr(objs, 'model', None) is Job:
67 # Django calls this branch for the jobs relation even when there are none; only record
68 # a count when there are actually jobs, so jobless objects don't get a spurious
69 # "0 jobs" row on the delete-confirmation page.
70 count = objs.count()
71 if count:
72 self.generic_relation_counts[Job] = self.generic_relation_counts.get(Job, 0) + count
73 return None
74 return super().collect(objs, source=source, *args, **kwargs)
77class CustomCollector(Collector):
78 """
79 Override Django's stock Collector to handle GenericRelations and ensure proper ordering of cascading deletions.
80 """
82 def collect(
83 self,
84 objs,
85 source=None,
86 nullable=False,
87 collect_related=True,
88 source_attr=None,
89 reverse_dependency=False,
90 keep_parents=False,
91 fail_on_restricted=True,
92 ):
93 # By default, Django will force the deletion of dependent objects before the parent only if the ForeignKey field
94 # is not nullable. We want to ensure proper ordering regardless, so if the ForeignKey has `on_delete=CASCADE`
95 # applied, we set `nullable` to False when calling `collect()`.
96 if objs and source and source_attr:
97 model = objs[0].__class__
98 field = model._meta.get_field(source_attr)
99 if field.remote_field.on_delete == CASCADE: 99 ↛ 102line 99 didn't jump to line 102 because the condition on line 99 was always true
100 nullable = False
102 super().collect(
103 objs,
104 source=source,
105 nullable=nullable,
106 collect_related=collect_related,
107 source_attr=source_attr,
108 reverse_dependency=reverse_dependency,
109 keep_parents=keep_parents,
110 fail_on_restricted=fail_on_restricted,
111 )
113 # Add GenericRelations to the dependency graph
114 processed_relations = set()
115 for _model, instances in list(self.data.items()):
116 for instance in instances:
117 # Get all GenericRelations for this model
118 for field in instance._meta.private_fields:
119 if isinstance(field, GenericRelation):
120 # Create a unique key for this relation
121 relation_key = f"{instance._meta.model_name}.{field.name}"
122 if relation_key in processed_relations:
123 continue
124 processed_relations.add(relation_key)
126 # Add the model that the generic relation points to as a dependency
127 self.add_dependency(field.related_model, instance, reverse_dependency=True)
130class DeleteMixin:
131 """
132 Mixin to override the model delete function to use our custom collector.
133 """
135 def delete(self, using=None, keep_parents=False):
136 """
137 Override delete to use our custom collector.
138 """
139 using = using or router.db_for_write(self.__class__, instance=self)
140 if self._get_pk_val() is None: 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true
141 raise ValueError(
142 _("{object_name} object can't be deleted because its {pk_attname} attribute is set to None.").format(
143 object_name=self._meta.object_name,
144 pk_attname=self._meta.pk.attname,
145 )
146 )
148 # Pass origin=self (matching Django's Model.delete) so signal receivers can tell that
149 # cascaded child objects are being deleted as part of deleting this object.
150 collector = CustomCollector(using=using, origin=self)
151 collector.collect([self], keep_parents=keep_parents)
153 return collector.delete()
155 delete.alters_data = True
157 @classmethod
158 def verify_mro(cls, instance):
159 """
160 Verify that this mixin is first in the MRO.
161 """
162 mro = instance.__class__.__mro__
163 if mro.index(cls) != 0:
164 raise RuntimeError(f"{cls.__name__} must be first in the MRO. Current MRO: {mro}")