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

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 if isinstance(obj, list) and all(isinstance(el, dict) for el in obj): 

69 ret = [] 

70 for el in obj: 

71 key, val = el.values() 

72 ret.append(str(val)) 

73 return ret 

74 return obj 

75 

76 def get_diff(self, other_version=None): 

77 if self.history_type == "-": 

78 return None 

79 version = other_version or self.prev_record 

80 if version: 

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

82 else: 

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

84 changes = list( 

85 filter( 

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

87 and x.field != "id" 

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

89 delta.changes, 

90 ) 

91 ) 

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

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

94 # we use a separate list containing the changed values 

95 newchanges = [] 

96 for change in changes: 

97 change = dataclasses.replace( 

98 change, 

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

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

101 ) 

102 newchanges.append(change) 

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

104 

105 

106class VersionMixin(models.Model): 

107 history = APISHistoricalRecords( 

108 inherit=True, 

109 bases=[ 

110 APISHistoryTableBase, 

111 ], 

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

113 ) 

114 __history_date = None 

115 

116 @property 

117 def _history_date(self): 

118 return self.__history_date or timezone.now() 

119 

120 @_history_date.setter 

121 def _history_date(self, value): 

122 self.__history_date = value 

123 pass 

124 

125 class Meta: 

126 abstract = True 

127 

128 def get_history_url(self): 

129 ct = ContentType.objects.get_for_model(self) 

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

131 

132 def _get_historical_relations(self): 

133 ret = [] 

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

135 from apis_core.relations.utils import relation_content_types 

136 

137 ct = ContentType.objects.get_for_model(self) 

138 

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

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

141 rel_history_models = [ 

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

143 ] 

144 

145 for model in rel_history_models: 

146 ret.append( 

147 model.history.filter( 

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

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

150 ).order_by("history_id") 

151 ) 

152 return ret 

153 

154 def get_history_data(self): 

155 data = [] 

156 prev_entry = None 

157 queries = self._get_historical_relations() 

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

159 

160 for entry in flatten_queries: 

161 if prev_entry is not None: 

162 if ( 

163 entry.history_date == prev_entry.history_date 

164 and entry.history_user_id == prev_entry.history_user_id 

165 ): 

166 entry.history_type = prev_entry.history_type 

167 data[-1] = entry 

168 prev_entry = entry 

169 continue 

170 data.append(entry) 

171 prev_entry = entry 

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

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

174 return data 

175 

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

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