Coverage for apis_core/apis_entities/models.py: 62%

1import functools 

2import re 

3 

4from django.conf import settings 

5from django.contrib.contenttypes.models import ContentType 

6from django.db.models.query import QuerySet 

7from django.db.models.signals import post_save 

8from django.dispatch import receiver 

9from django.urls import NoReverseMatch, reverse 

10 

11from apis_core.apis_entities import signals 

12from apis_core.apis_metainfo.models import RootObject, Uri 

13 

14NEXT_PREV = getattr(settings, "APIS_NEXT_PREV", True) 

15 

16 

17class AbstractEntity(RootObject): 

18 """ 

19 Abstract super class which encapsulates common logic between the 

20 different entity kinds and provides various methods relating to either 

21 all or one specific entity kind. 

22 

23 Most of the class methods are designed to be used in the subclass as they 

24 are considering contexts which depend on the subclass entity type. 

25 So they are to be understood in that dynamic context. 

26 """ 

27 

28 class Meta: 

29 abstract = True 

30 

31 def __init__(self, *args, **kwargs): 

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

33 

34 @classmethod 

35 def get_or_create_uri(cls, uri): 

36 uri = str(uri) 

37 try: 

38 if re.match(r"^[0-9]*$", uri): 

39 p = cls.objects.get(pk=uri) 

40 else: 

41 p = cls.objects.get(uri__uri=uri) 

42 return p 

43 except Exception as e: 

44 print("Found no object corresponding to given uri." + e) 

45 return False 

46 

47 # TODO 

48 @classmethod 

49 def get_entity_list_filter(cls): 

50 return None 

51 

52 @functools.cached_property 

53 def get_prev_id(self): 

54 if NEXT_PREV: 

55 prev_instance = ( 

56 type(self) 

57 .objects.filter(id__lt=self.id) 

58 .order_by("-id") 

59 .only("id") 

60 .first() 

61 ) 

62 if prev_instance is not None: 

63 return prev_instance.id 

64 return False 

65 

66 @functools.cached_property 

67 def get_next_id(self): 

68 if NEXT_PREV: 

69 next_instance = ( 

70 type(self) 

71 .objects.filter(id__gt=self.id) 

72 .order_by("id") 

73 .only("id") 

74 .first() 

75 ) 

76 if next_instance is not None: 

77 return next_instance.id 

78 return False 

79 

80 def get_duplicate_url(self): 

81 entity = self.__class__.__name__.lower() 

82 return reverse( 

83 "apis_core:apis_entities:generic_entities_duplicate_view", 

84 kwargs={"contenttype": entity, "pk": self.id}, 

85 ) 

86 

87 def get_merge_view_url(self): 

88 entity = self.__class__.__name__.lower() 

89 return reverse( 

90 "apis_core:apis_entities:generic_entities_merge_view", 

91 kwargs={"contenttype": entity, "pk": self.id}, 

92 ) 

93 

94 def merge_start_date_written(self, other): 

95 self.start_date_written = self.start_date_written or other.start_date_written 

96 

97 def merge_end_date_written(self, other): 

98 self.end_date_written = self.end_date_written or other.end_date_written 

99 

100 def merge_with(self, entities): 

101 if self in entities: 

102 entities.remove(self) 

103 origin = self.__class__ 

104 signals.pre_merge_with.send(sender=origin, instance=self, entities=entities) 

105 

106 # TODO: check if these imports can be put to top of module without 

107 # causing circular import issues. 

108 from apis_core.apis_metainfo.models import Uri 

109 

110 e_a = type(self).__name__ 

111 self_model_class = ContentType.objects.get(model__iexact=e_a).model_class() 

112 if isinstance(entities, int): 

113 entities = self_model_class.objects.get(pk=entities) 

114 if not isinstance(entities, list) and not isinstance(entities, QuerySet): 

115 entities = [entities] 

116 entities = [ 

117 self_model_class.objects.get(pk=ent) if isinstance(ent, int) else ent 

118 for ent in entities 

119 ] 

120 for ent in entities: 

121 e_b = type(ent).__name__ 

122 if e_a != e_b: 

123 continue 

124 for f in ent._meta.local_many_to_many: 

125 if not f.name.endswith("_set"): 

126 sl = list(getattr(self, f.name).all()) 

127 for s in getattr(ent, f.name).all(): 

128 if s not in sl: 

129 getattr(self, f.name).add(s) 

130 Uri.objects.filter(root_object=ent).update(root_object=self) 

131 

132 for ent in entities: 

133 self.merge_fields(ent) 

134 

135 signals.post_merge_with.send(sender=origin, instance=self, entities=entities) 

136 

137 for ent in entities: 

138 ent.delete() 

139 

140 def get_serialization(self): 

141 from apis_core.apis_entities.serializers_generic import EntitySerializer 

142 

143 return EntitySerializer(self).data 

144 

145 

146@receiver(post_save, dispatch_uid="create_default_uri") 

147def create_default_uri(sender, instance, created, raw, using, update_fields, **kwargs): 

148 create_default_uri = getattr(settings, "CREATE_DEFAULT_URI", True) 

149 skip_default_uri = getattr(instance, "skip_default_uri", False) 

150 if create_default_uri and not skip_default_uri: 

151 if isinstance(instance, AbstractEntity) and created: 

152 base = getattr(settings, "APIS_BASE_URI", "https://example.org").strip("/") 

153 try: 

154 route = reverse("GetEntityGenericRoot", kwargs={"pk": instance.pk}) 

155 except NoReverseMatch: 

156 route = reverse( 

157 "apis_core:GetEntityGeneric", kwargs={"pk": instance.pk} 

158 ) 

159 uri = f"{base}{route}" 

160 Uri.objects.create(uri=uri, root_object=instance)