Coverage for apis_core/history/models.py: 54%

1import inspect 

2from typing import Any 

3 

4import django 

5from django.conf import settings 

6from django.contrib.contenttypes.models import ContentType 

7from django.core.exceptions import AppRegistryNotReady 

8from django.db import models 

9from django.db.models import Q, UniqueConstraint 

10from django.db.models.functions import Lower 

11from django.urls import reverse 

12from django.utils import timezone 

13from simple_history import utils 

14from simple_history.models import HistoricalRecords 

15 

16from apis_core.generic.abc import GenericModel 

17 

18 

19class APISHistoricalRecords(HistoricalRecords, GenericModel): 

20 def get_m2m_fields_from_model(self, model): 

21 # Change the original simple history function to also return m2m fields 

22 m2m_fields = [] 

23 try: 

24 for field in inspect.getmembers(model): 

25 if isinstance( 

26 field[1], 

27 django.db.models.fields.related_descriptors.ManyToManyDescriptor, 

28 ): 

29 m2m_fields.append(getattr(model, field[0]).field) 

30 except AppRegistryNotReady: 

31 pass 

32 return m2m_fields 

33 

34 def get_prev_record(self): 

35 """ 

36 Get the previous history record for the instance. `None` if first. 

37 """ 

38 history = utils.get_history_manager_from_history(self) 

39 return ( 

40 history.filter(history_date__lt=self.history_date) 

41 .order_by("history_date") 

42 .first() 

43 ) 

44 

45 

46class APISHistoryTableBase(models.Model, GenericModel): 

47 version_tag = models.CharField(max_length=255, blank=True, null=True) 

48 

49 class Meta: 

50 abstract = True 

51 constraints = [ 

52 UniqueConstraint( 

53 fields=["id", Lower("version_tag")], name="id_version_tag_unique" 

54 ), 

55 ] 

56 

57 def set_version_tag(self, tag: str): 

58 self.version_tag = tag 

59 self.save() 

60 

61 def get_absolute_url(self): 

62 ct = ContentType.objects.get_for_model(self) 

63 return reverse("apis_core:generic:detail", args=[ct, self.history_id]) 

64 

65 def get_diff(self, other_version=None): 

66 if self.history_type == "-": 

67 return None 

68 version = other_version or self.prev_record 

69 if version: 

70 delta = self.diff_against(version) 

71 else: 

72 delta = self.diff_against(self.__class__()) 

73 changes = list( 

74 filter( 

75 lambda x: (x.new != "" or x.old is not None) 

76 and x.field != "id" 

77 and not x.field.endswith("_ptr"), 

78 delta.changes, 

79 ) 

80 ) 

81 return sorted(changes, key=lambda change: change.field) 

82 

83 

84class VersionMixin(models.Model): 

85 history = APISHistoricalRecords( 

86 inherit=True, 

87 bases=[ 

88 APISHistoryTableBase, 

89 ], 

90 custom_model_name=lambda x: f"Version{x}", 

91 ) 

92 __history_date = None 

93 

94 @property 

95 def _history_date(self): 

96 return self.__history_date or timezone.now() 

97 

98 @_history_date.setter 

99 def _history_date(self, value): 

100 self.__history_date = value 

101 pass 

102 

103 class Meta: 

104 abstract = True 

105 

106 def get_history_url(self): 

107 ct = ContentType.objects.get_for_model(self) 

108 return reverse("apis_core:history:history", args=[ct, self.id]) 

109 

110 def get_create_version_url(self): 

111 ct = ContentType.objects.get_for_model(self) 

112 return reverse("apis_core:history:add_new_history_version", args=[ct, self.id]) 

113 

114 def _get_historical_relations(self): 

115 ret = [] 

116 if "apis_core.relations" in settings.INSTALLED_APPS: 

117 from apis_core.relations.utils import relation_content_types 

118 

119 ct = ContentType.objects.get_for_model(self) 

120 

121 rel_content_types = relation_content_types(any_model=type(self)) 

122 rel_models = [ct.model_class() for ct in rel_content_types] 

123 rel_history_models = [ 

124 model for model in rel_models if issubclass(model, VersionMixin) 

125 ] 

126 

127 for model in rel_history_models: 

128 ret.append( 

129 model.history.filter( 

130 Q(subj_object_id=self.id, subj_content_type=ct) 

131 | Q(obj_object_id=self.id, obj_content_type=ct) 

132 ).order_by("history_id") 

133 ) 

134 return ret 

135 

136 def get_history_data(self): 

137 data = [] 

138 prev_entry = None 

139 queries = self._get_historical_relations() 

140 flatten_queries = [entry for query in queries for entry in query] 

141 

142 for entry in flatten_queries: 

143 if prev_entry is not None: 

144 if ( 

145 entry.history_date == prev_entry.history_date 

146 and entry.history_user_id == prev_entry.history_user_id 

147 ): 

148 entry.history_type = prev_entry.history_type 

149 data[-1] = entry 

150 prev_entry = entry 

151 continue 

152 data.append(entry) 

153 prev_entry = entry 

154 data += [x for x in self.history.all()] 

155 data = sorted(data, key=lambda x: x.history_date, reverse=True) 

156 return data 

157 

158 def __init__(self, *args: Any, **kwargs: Any) -> None: 

159 super().__init__(*args, **kwargs)