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

1import inspect 

2from datetime import datetime 

3from typing import Any 

4 

5import django 

6from django.conf import settings 

7from django.contrib.contenttypes.models import ContentType 

8from django.core.exceptions import AppRegistryNotReady 

9from django.db import models 

10from django.db.models import Q, UniqueConstraint 

11from django.db.models.functions import Lower 

12from django.urls import reverse 

13from django.utils import timezone 

14from simple_history import utils 

15from simple_history.models import HistoricalRecords 

16 

17from apis_core.apis_metainfo.models import RootObject 

18from apis_core.generic.abc import GenericModel 

19 

20 

21class APISHistoricalRecords(HistoricalRecords, GenericModel): 

22 def get_m2m_fields_from_model(self, model): 

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

24 m2m_fields = [] 

25 try: 

26 for field in inspect.getmembers(model): 

27 if isinstance( 

28 field[1], 

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

30 ): 

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

32 except AppRegistryNotReady: 

33 pass 

34 return m2m_fields 

35 

36 def get_prev_record(self): 

37 """ 

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

39 """ 

40 history = utils.get_history_manager_from_history(self) 

41 return ( 

42 history.filter(history_date__lt=self.history_date) 

43 .order_by("history_date") 

44 .first() 

45 ) 

46 

47 

48class APISHistoryTableBase(models.Model, GenericModel): 

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

50 

51 class Meta: 

52 abstract = True 

53 constraints = [ 

54 UniqueConstraint( 

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

56 ), 

57 ] 

58 

59 def get_triples_for_version( 

60 self, 

61 only_latest: bool = True, 

62 history_date: datetime = None, 

63 filter_for_triples: bool = True, 

64 ): 

65 """returns all triples for a specific version of a model instance. 

66 If only_latest is True, only the latest version of a triple is returned.""" 

67 from apis_core.apis_relations.models import TempTriple 

68 

69 if not isinstance(self.instance, RootObject): 

70 return TempTriple.objects.none() 

71 

72 if history_date is None: 

73 filter_date = ( 

74 self.next_record.history_date if self.next_record else datetime.now() 

75 ) 

76 else: 

77 filter_date = history_date 

78 triples = TempTriple.history.filter( 

79 Q(subj=self.instance) | Q(obj=self.instance), history_date__lte=filter_date 

80 ) 

81 if self.version_tag and filter_for_triples: 

82 triples = triples.filter( 

83 Q(version_tag=self.version_tag) 

84 | Q(version_tag__contains=f"{self.version_tag},") 

85 ) 

86 if only_latest: 

87 triples = triples.latest_of_each() 

88 if filter_for_triples: 

89 triples = triples.exclude(history_type="-") 

90 return triples 

91 

92 def set_version_tag(self, tag: str, include_triples: bool = True): 

93 self.version_tag = tag 

94 if include_triples: 

95 triples = self.get_triples_for_version(filter_for_triples=False) 

96 for triple in triples: 

97 if triple.version_tag is None: 

98 triple.version_tag = tag 

99 else: 

100 triple.version_tag = f"{triple.version_tag},{tag},".replace( 

101 ",,", "," 

102 ) 

103 triple.save() 

104 self.save() 

105 

106 def get_absolute_url(self): 

107 ct = ContentType.objects.get_for_model(self) 

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

109 

110 def get_diff(self, other_version=None): 

111 if self.history_type == "-": 

112 return None 

113 version = other_version or self.prev_record 

114 if version: 

115 delta = self.diff_against(version) 

116 else: 

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

118 changes = list( 

119 filter( 

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

121 and x.field != "id" 

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

123 delta.changes, 

124 ) 

125 ) 

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

127 

128 

129class VersionMixin(models.Model): 

130 history = APISHistoricalRecords( 

131 inherit=True, 

132 bases=[ 

133 APISHistoryTableBase, 

134 ], 

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

136 ) 

137 __history_date = None 

138 

139 @property 

140 def _history_date(self): 

141 return self.__history_date or timezone.now() 

142 

143 @_history_date.setter 

144 def _history_date(self, value): 

145 self.__history_date = value 

146 pass 

147 

148 class Meta: 

149 abstract = True 

150 

151 def get_history_url(self): 

152 ct = ContentType.objects.get_for_model(self) 

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

154 

155 def get_create_version_url(self): 

156 ct = ContentType.objects.get_for_model(self) 

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

158 

159 def _get_historical_relations(self): 

160 ret = [] 

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

162 from apis_core.relations.utils import relation_content_types 

163 

164 ct = ContentType.objects.get_for_model(self) 

165 

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

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

168 rel_history_models = [ 

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

170 ] 

171 

172 for model in rel_history_models: 

173 ret.append( 

174 model.history.filter( 

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

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

177 ).order_by("history_id") 

178 ) 

179 return ret 

180 

181 def _get_historical_triples(self): 

182 # TODO: this is a workaround to filter out Triple history entries and leave 

183 # TempTriple entries only. Fix when we switch to new relations model. 

184 ret = [] 

185 if "apis_core.apis_relations" in settings.INSTALLED_APPS: 

186 from apis_core.apis_relations.models import TempTriple 

187 

188 ret.append( 

189 TempTriple.history.filter(Q(subj=self) | Q(obj=self)).order_by( 

190 "history_id" 

191 ) 

192 ) 

193 return ret 

194 

195 def get_history_data(self): 

196 data = [] 

197 prev_entry = None 

198 queries = self._get_historical_relations() + self._get_historical_triples() 

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

200 

201 for entry in flatten_queries: 

202 if prev_entry is not None: 

203 if ( 

204 entry.history_date == prev_entry.history_date 

205 and entry.history_user_id == prev_entry.history_user_id 

206 ): 

207 entry.history_type = prev_entry.history_type 

208 data[-1] = entry 

209 prev_entry = entry 

210 continue 

211 data.append(entry) 

212 prev_entry = entry 

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

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

215 return data 

216 

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

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