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

1import dataclasses 

2import inspect 

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 

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(GenericModel, models.Model): 

47 class Meta: 

48 abstract = True 

49 

50 def get_absolute_url(self): 

51 ct = ContentType.objects.get_for_model(self) 

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

53 

54 def get_reset_url(self): 

55 ct = ContentType.objects.get_for_model(self) 

56 return reverse("apis_core:history:reset", args=[ct, self.history_id]) 

57 

58 def _flatten_modelchange_many_to_many(self, obj: Any) -> Any: 

59 """ 

60 The `.diff_against` method of the `HistoricalChanges` model represent 

61 changes in many-to-many fields as lists of dicts. Those dicts contain 

62 a mapping of the model names to the model instances (i.e. 

63 {'person': <Person1>, 'profession': <Profession2>}). 

64 To make this a bit more readable, we flatten the dicts to one dict 

65 containing a mapping between the model instance string representation 

66 and the string representations of the connected model instances. 

67 """ 

68 ret = [] 

69 for el in obj: 

70 key, val = el.values() 

71 ret.append(str(val)) 

72 return ret 

73 

74 def get_diff(self, other_version=None): 

75 if self.history_type == "-": 

76 return None 

77 version = other_version or self.prev_record 

78 if version: 

79 delta = self.diff_against(version, foreign_keys_are_objs=True) 

80 else: 

81 delta = self.diff_against(self.__class__(), foreign_keys_are_objs=True) 

82 changes = list( 

83 filter( 

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

85 and x.field != "id" 

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

87 delta.changes, 

88 ) 

89 ) 

90 # The changes consist of `ModelChange` classes, which are frozen 

91 # To flatten the many-to-many representation of those ModelChanges 

92 # we use a separate list containing the changed values 

93 newchanges = [] 

94 m2m_fields = {field.name for field in self.instance_type._meta.many_to_many} 

95 for change in changes: 

96 # run flattening only on m2m fields 

97 if change.field in m2m_fields: 

98 change = dataclasses.replace( 

99 change, 

100 old=self._flatten_modelchange_many_to_many(change.old), 

101 new=self._flatten_modelchange_many_to_many(change.new), 

102 ) 

103 newchanges.append(change) 

104 return sorted(newchanges, key=lambda change: change.field) 

105 

106 

107class VersionMixin(models.Model): 

108 history = APISHistoricalRecords( 

109 inherit=True, 

110 bases=[ 

111 APISHistoryTableBase, 

112 ], 

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

114 ) 

115 __history_date = None 

116 

117 @property 

118 def _history_date(self): 

119 return self.__history_date or timezone.now() 

120 

121 @_history_date.setter 

122 def _history_date(self, value): 

123 self.__history_date = value 

124 pass 

125 

126 class Meta: 

127 abstract = True 

128 

129 def get_history_url(self): 

130 ct = ContentType.objects.get_for_model(self) 

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

132 

133 def _get_historical_relations(self): 

134 ret = [] 

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

136 from apis_core.relations.utils import relation_content_types 

137 

138 ct = ContentType.objects.get_for_model(self) 

139 

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

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

142 rel_history_models = [ 

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

144 ] 

145 

146 for model in rel_history_models: 

147 ret.append( 

148 model.history.filter( 

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

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

151 ).order_by("history_id") 

152 ) 

153 return ret 

154 

155 def get_history_data(self): 

156 data = [] 

157 prev_entry = None 

158 queries = self._get_historical_relations() 

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

160 

161 for entry in flatten_queries: 

162 if prev_entry is not None: 

163 if ( 

164 entry.history_date == prev_entry.history_date 

165 and entry.history_user_id == prev_entry.history_user_id 

166 ): 

167 entry.history_type = prev_entry.history_type 

168 data[-1] = entry 

169 prev_entry = entry 

170 continue 

171 data.append(entry) 

172 prev_entry = entry 

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

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

175 return data 

176 

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

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