Coverage for extras/management/commands/reindex.py: 54%

1from django.contrib.contenttypes.models import ContentType 

2from django.core.management.base import BaseCommand, CommandError 

3from django.utils.translation import gettext as _ 

4 

5from netbox.registry import registry 

6from netbox.search.backends import search_backend 

7 

8 

9class Command(BaseCommand): 

10 help = 'Reindex objects for search' 

11 

12 def add_arguments(self, parser): 

13 parser.add_argument( 

14 'args', 

15 metavar='app_label[.ModelName]', 

16 nargs='*', 

17 help='One or more apps or models to reindex', 

18 ) 

19 parser.add_argument( 

20 '--lazy', 

21 action='store_true', 

22 help="For each model, reindex objects only if no cache entries already exist" 

23 ) 

24 

25 def _get_indexers(self, *model_names): 

26 indexers = {} 

27 

28 # No models specified; pull in all registered indexers 

29 if not model_names: 29 ↛ 35line 29 didn't jump to line 35 because the condition on line 29 was always true

30 for idx in registry['search'].values(): 

31 indexers[idx.model] = idx 

32 

33 # Return only indexers for the specified models 

34 else: 

35 for label in model_names: 

36 labels = label.lower().split('.') 

37 

38 # Label specifies an exact model 

39 if len(labels) == 2: 

40 app_label, model_name = labels 

41 try: 

42 idx = registry['search'][f'{app_label}.{model_name}'] 

43 indexers[idx.model] = idx 

44 except KeyError: 

45 raise CommandError(f"No indexer registered for {label}") 

46 

47 # Label specifies all the models of an app 

48 elif len(labels) == 1: 

49 app_label = labels[0] + '.' 

50 for indexer_label, idx in registry['search'].items(): 

51 if indexer_label.startswith(app_label): 

52 indexers[idx.model] = idx 

53 

54 else: 

55 raise CommandError( 

56 f"Invalid model: {label}. Model names must be in the format <app_label> or " 

57 f"<app_label>.<model_name>." 

58 ) 

59 

60 return indexers 

61 

62 def handle(self, *model_labels, **kwargs): 

63 

64 # Determine which models to reindex 

65 indexers = self._get_indexers(*model_labels) 

66 if not indexers: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 raise CommandError(_("No indexers found!")) 

68 self.stdout.write(f'Reindexing {len(indexers)} models.') 

69 

70 # Clear cached values for the specified models (if not being lazy) 

71 if not kwargs['lazy']: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true

72 if model_labels: 

73 content_types = [ContentType.objects.get_for_model(model) for model in indexers.keys()] 

74 else: 

75 content_types = None 

76 

77 self.stdout.write('Clearing cached values... ', ending='') 

78 self.stdout.flush() 

79 deleted_count = search_backend.clear(object_types=content_types) 

80 self.stdout.write(f'{deleted_count} entries deleted.') 

81 

82 # Index models 

83 self.stdout.write('Indexing models') 

84 for model, idx in indexers.items(): 

85 app_label = model._meta.app_label 

86 model_name = model._meta.model_name 

87 self.stdout.write(f' {app_label}.{model_name}... ', ending='') 

88 self.stdout.flush() 

89 

90 if kwargs['lazy']: 90 ↛ 96line 90 didn't jump to line 96 because the condition on line 90 was always true

91 content_type = ContentType.objects.get_for_model(model) 

92 if cached_count := search_backend.count(object_types=[content_type]): 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 self.stdout.write(f'Skipping (found {cached_count} existing).') 

94 continue 

95 

96 i = search_backend.cache(model.objects.iterator(), remove_existing=False) 

97 if i: 

98 self.stdout.write(f'{i} entries cached.') 

99 else: 

100 self.stdout.write('No objects found.') 

101 

102 msg = 'Completed.' 

103 if total_count := search_backend.size: 103 ↛ 105line 103 didn't jump to line 105 because the condition on line 103 was always true

104 msg += f' Total entries: {total_count}' 

105 self.stdout.write(msg, self.style.SUCCESS)