Coverage for extras/management/commands/renaturalize.py: 0%

1from django.apps import apps 

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

3 

4from utilities.fields import NaturalOrderingField 

5from utilities.querysets import chunked_update 

6 

7 

8class Command(BaseCommand): 

9 help = "Recalculate natural ordering values for the specified models" 

10 

11 def add_arguments(self, parser): 

12 parser.add_argument( 

13 'args', metavar='app_label.ModelName', nargs='*', 

14 help='One or more specific models (each prefixed with its app_label) to renaturalize', 

15 ) 

16 

17 def _get_models(self, names): 

18 """ 

19 Compile a list of models to be renaturalized. If no names are specified, all models which have one or more 

20 NaturalOrderingFields will be included. 

21 """ 

22 models = [] 

23 

24 if names: 

25 # Collect all NaturalOrderingFields present on the specified models 

26 for name in names: 

27 try: 

28 app_label, model_name = name.split('.') 

29 except ValueError: 

30 raise CommandError( 

31 f"Invalid format: {name}. Models must be specified in the form app_label.ModelName." 

32 ) 

33 try: 

34 app_config = apps.get_app_config(app_label) 

35 except LookupError as e: 

36 raise CommandError(str(e)) 

37 try: 

38 model = app_config.get_model(model_name) 

39 except LookupError: 

40 raise CommandError(f"Unknown model: {app_label}.{model_name}") 

41 fields = [ 

42 field for field in model._meta.concrete_fields if type(field) is NaturalOrderingField 

43 ] 

44 if not fields: 

45 raise CommandError( 

46 f"Invalid model: {app_label}.{model_name} does not employ natural ordering" 

47 ) 

48 models.append( 

49 (model, fields) 

50 ) 

51 

52 else: 

53 # Find *all* models with NaturalOrderingFields 

54 for app_config in apps.get_app_configs(): 

55 for model in app_config.models.values(): 

56 fields = [ 

57 field for field in model._meta.concrete_fields if type(field) is NaturalOrderingField 

58 ] 

59 if fields: 

60 models.append( 

61 (model, fields) 

62 ) 

63 

64 return models 

65 

66 def handle(self, *args, **options): 

67 

68 models = self._get_models(args) 

69 

70 if options['verbosity']: 

71 self.stdout.write(f"Renaturalizing {len(models)} models.") 

72 

73 for model, fields in models: 

74 for field in fields: 

75 target_field = field.target_field 

76 naturalize = field.naturalize_function 

77 count = 0 

78 

79 # Print the model and field name 

80 if options['verbosity']: 

81 self.stdout.write( 

82 f"{model._meta.label}.{field.target_field} ({field.name})... ", 

83 ending='\n' if options['verbosity'] >= 2 else '' 

84 ) 

85 self.stdout.flush() 

86 

87 # Find all unique values for the field 

88 queryset = model.objects.values_list(target_field, flat=True).order_by(target_field).distinct() 

89 for value in queryset: 

90 naturalized_value = naturalize(value, max_length=field.max_length) 

91 

92 if options['verbosity'] >= 2: 

93 self.stdout.write(f" {value} -> {naturalized_value}", ending='') 

94 self.stdout.flush() 

95 

96 # Update each unique field value in bulk 

97 changed = chunked_update(model.objects.filter(name=value), **{field.name: naturalized_value}) 

98 

99 if options['verbosity'] >= 2: 

100 self.stdout.write(f" ({changed})") 

101 count += changed 

102 

103 # Print the total count of alterations for the field 

104 if options['verbosity'] >= 2: 

105 self.stdout.write(self.style.SUCCESS( 

106 f"{count} {model._meta.verbose_name_plural} updated ({queryset.count()} unique values)" 

107 )) 

108 elif options['verbosity']: 

109 self.stdout.write(self.style.SUCCESS(str(count))) 

110 

111 if options['verbosity']: 

112 self.stdout.write(self.style.SUCCESS("Done."))