NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-12-09 09:59:37 -06:00
Code Issues Actions 13 Packages Projects Releases Wiki Activity

fix(utilities): Ensure unique signal handlers for counter models

Browse Source 
Updates `connect_counters` to prevent duplicate signal handlers by
using consistent `dispatch_uid` values per sender. Adds a check to
avoid reconnecting models already processed during registration.

Fixes #20697
This commit is contained in:
Martin Hauser 2025-10-28 17:06:25 +01:00 committed by Jeremy Stretch
parent 6a21459ccc
commit 70bc1c226a
1 changed files with 20 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
netbox/utilities/counters.py
View File

@ -77,7 +77,7 @@ def post_delete_receiver(sender, instance, origin, **kwargs):
parent_pk = getattr(instance, field_name, None)
# Decrement the parent's counter by one
if parent_pk is not None and not hasattr(instance, "_previously_removed"):
if parent_pk is not None and not hasattr(instance, '_previously_removed'):
update_counter(parent_model, parent_pk, counter_name, -1)
@ -87,38 +87,48 @@ def post_delete_receiver(sender, instance, origin, **kwargs):
def connect_counters(*models):
"""
Register counter fields and connect post_save & post_delete signal handlers for the affected models.
Register counter fields and connect signal handlers for their child models.
Ensures exactly one receiver per child (sender), even when multiple counters
reference the same sender (e.g., Device).
"""
for model in models:
connected = set() # child models we've already connected
for model in models:
# Find all CounterCacheFields on the model
counter_fields = [
field for field in model._meta.get_fields() if type(field) is CounterCacheField
]
counter_fields = [field for field in model._meta.get_fields() if isinstance(field, CounterCacheField)]
for field in counter_fields:
to_model = apps.get_model(field.to_model_name)
# Register the counter in the registry
change_tracking_fields = registry['counter_fields'][to_model]
change_tracking_fields[f"{field.to_field_name}_id"] = field.name
change_tracking_fields[f'{field.to_field_name}_id'] = field.name
# Connect signals once per child model
if to_model in connected:
continue
# Ensure dispatch_uid is unique per model (sender), not per field
uid_base = f'countercache.{to_model._meta.label_lower}'
# Connect the post_save and post_delete handlers
post_save.connect(
post_save_receiver,
sender=to_model,
weak=False,
dispatch_uid=f'{model._meta.label}.{field.name}'
dispatch_uid=f'{uid_base}.post_save',
)
pre_delete.connect(
pre_delete_receiver,
sender=to_model,
weak=False,
dispatch_uid=f'{model._meta.label}.{field.name}'
dispatch_uid=f'{uid_base}.pre_delete',
)
post_delete.connect(
post_delete_receiver,
sender=to_model,
weak=False,
dispatch_uid=f'{model._meta.label}.{field.name}'
dispatch_uid=f'{uid_base}.post_delete',
)
connected.add(to_model)