Coverage for extras/managers.py: 38%

1from django.db import router 

2from django.db.models import signals 

3from taggit.managers import TaggableManager, _TaggableManager 

4from taggit.utils import require_instance_manager 

5 

6__all__ = ( 

7 'NetBoxTaggableManager', 

8 'NetBoxTaggableManagerField', 

9) 

10 

11 

12class NetBoxTaggableManager(_TaggableManager): 

13 """ 

14 Extends taggit's _TaggableManager to: 

15 

16 * Replace the per-tag get_or_create loop in add() with a single bulk_create() call, reducing 

17 SQL queries from O(N) to O(1) when assigning tags. 

18 * Implement set_base(), the M2M assignment entry point Django's deserializer calls. 

19 """ 

20 

21 @require_instance_manager 

22 def add(self, *tags, through_defaults=None, tag_kwargs=None, **kwargs): 

23 self._remove_prefetched_objects() 

24 if tag_kwargs is None: 

25 tag_kwargs = {} 

26 db = router.db_for_write(self.through, instance=self.instance) 

27 

28 tag_objs = self._to_tag_model_instances(tags, tag_kwargs) 

29 new_ids = {t.pk for t in tag_objs} 

30 

31 # Determine which tags are not already assigned to this object 

32 lookup = self._lookup_kwargs() 

33 vals = set( 

34 self.through._default_manager.using(db) 

35 .values_list("tag_id", flat=True) 

36 .filter(**lookup, tag_id__in=new_ids) 

37 ) 

38 new_ids -= vals 

39 

40 if not new_ids: 

41 return 

42 

43 signals.m2m_changed.send( 

44 sender=self.through, 

45 action="pre_add", 

46 instance=self.instance, 

47 reverse=False, 

48 model=self.through.tag_model(), 

49 pk_set=new_ids, 

50 using=db, 

51 ) 

52 

53 # Use a single bulk INSERT instead of one get_or_create per tag. 

54 self.through._default_manager.using(db).bulk_create( 

55 [ 

56 self.through(tag=tag, **lookup, **(through_defaults or {})) 

57 for tag in tag_objs 

58 if tag.pk in new_ids 

59 ], 

60 ignore_conflicts=True, 

61 ) 

62 

63 signals.m2m_changed.send( 

64 sender=self.through, 

65 action="post_add", 

66 instance=self.instance, 

67 reverse=False, 

68 model=self.through.tag_model(), 

69 pk_set=new_ids, 

70 using=db, 

71 ) 

72 

73 @require_instance_manager 

74 def set_base(self, objs, *, clear=False, through_defaults=None, raw=False): 

75 # Django's deserializer assigns M2M data through this method, passing primary keys; 

76 # taggit's set() takes only Tag instances or names. Keys which match no tag are passed 

77 # through for the database to reject, as ManyRelatedManager.set_base() does. 

78 tag_model = self.through.tag_model() 

79 if pks := [obj for obj in objs if not isinstance(obj, (tag_model, str))]: 

80 db = router.db_for_write(self.through, instance=self.instance) 

81 tags = tag_model._default_manager.using(db).in_bulk(pks) 

82 objs = [ 

83 obj if isinstance(obj, (tag_model, str)) else tags.get(obj) or tag_model(pk=obj) 

84 for obj in objs 

85 ] 

86 return self.set(objs, clear=clear, through_defaults=through_defaults) 

87 

88 

89class NetBoxTaggableManagerField(TaggableManager): 

90 """ 

91 Subclass of taggit's TaggableManager that interpolates `%(app_label)s` and `%(class)s` in 

92 `related_name`. taggit's contribute_to_class() bypasses Django's RelatedField, which is what 

93 normally performs this substitution, so without this two taggable models that share a class 

94 name (e.g. from different plugins) collide on Tag's reverse accessor. 

95 """ 

96 def contribute_to_class(self, cls, name): 

97 super().contribute_to_class(cls, name) 

98 if not cls._meta.abstract and self.remote_field.related_name: 

99 self.remote_field.related_name = self.remote_field.related_name % { 

100 'class': cls.__name__.lower(), 

101 'app_label': cls._meta.app_label.lower(), 

102 } 

103 

104 def deconstruct(self): 

105 # Emit the upstream taggit path and omit related_name so existing migrations remain 

106 # equivalent and no AlterField is produced for every TagsMixin consumer. related_name 

107 # has no effect on the database schema; it is reapplied on model load. Only safe while 

108 # this subclass adds no field attributes that affect schema — if that changes, restore 

109 # the real path so migrations capture the diff. 

110 name, _path, args, kwargs = super().deconstruct() 

111 kwargs.pop('related_name', None) 

112 return name, 'taggit.managers.TaggableManager', args, kwargs