Coverage for apis_core/apis_metainfo/models.py: 56%

1import logging 

2 

3from django.conf import settings 

4from django.contrib.contenttypes.models import ContentType 

5from django.core.exceptions import ImproperlyConfigured, ValidationError 

6from django.db import models 

7from django.db.models.fields.related import ForeignKey, ManyToManyField 

8from django.db.models.fields.related_descriptors import ForwardManyToOneDescriptor 

9from django.forms import model_to_dict 

10from model_utils.managers import InheritanceManager 

11 

12from apis_core.apis_metainfo import signals 

13from apis_core.generic.abc import GenericModel 

14from apis_core.utils import rdf 

15from apis_core.utils.normalize import clean_uri 

16 

17logger = logging.getLogger(__name__) 

18 

19 

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

21 

22 

23class RootObject(GenericModel, models.Model): 

24 """ 

25 The very root thing that can exist in a given ontology. Several classes inherit from it. 

26 By having one overarching super class we gain the advantage of unique identifiers. 

27 """ 

28 

29 # self_contenttype: a foreign key to the respective contenttype comes in handy when querying for 

30 # triples where the subject's or object's contenttype must be respected (e.g. get all triples 

31 # where the subject is a Person) 

32 self_contenttype = models.ForeignKey( 

33 ContentType, 

34 on_delete=models.deletion.CASCADE, 

35 null=True, 

36 blank=True, 

37 editable=False, 

38 ) 

39 objects = models.Manager() 

40 objects_inheritance = InheritanceManager() 

41 

42 def save(self, *args, **kwargs): 

43 self.self_contenttype = ContentType.objects.get_for_model(self) 

44 super().save(*args, **kwargs) 

45 

46 def duplicate(self): 

47 origin = self.__class__ 

48 signals.pre_duplicate.send(sender=origin, instance=self) 

49 # usually, copying instances would work like 

50 # https://docs.djangoproject.com/en/4.2/topics/db/queries/#copying-model-instances 

51 # but we are working with abstract classes, 

52 # so we have to do it by hand using model_to_dict:( 

53 objdict = model_to_dict(self) 

54 

55 # remove unique fields from dict representation 

56 unique_fields = [field for field in self._meta.fields if field.unique] 

57 for field in unique_fields: 

58 logger.info(f"Duplicating {self}: ignoring unique field {field.name}") 

59 objdict.pop(field.name, None) 

60 

61 # remove related fields from dict representation 

62 related_fields = [ 

63 field for field in self._meta.get_fields() if field.is_relation 

64 ] 

65 for field in related_fields: 

66 objdict.pop(field.name, None) 

67 

68 newobj = type(self).objects.create(**objdict) 

69 

70 for field in related_fields: 

71 # we are not using `isinstance` because we want to 

72 # differentiate between different levels of inheritance 

73 if type(field) is ForeignKey: 

74 setattr(newobj, field.name, getattr(self, field.name)) 

75 if type(field) is ManyToManyField: 

76 objfield = getattr(newobj, field.name) 

77 values = getattr(self, field.name).all() 

78 objfield.set(values) 

79 

80 newobj.save() 

81 signals.post_duplicate.send(sender=origin, instance=self, duplicate=newobj) 

82 return newobj 

83 

84 duplicate.alters_data = True 

85 

86 

87class InheritanceForwardManyToOneDescriptor(ForwardManyToOneDescriptor): 

88 def get_queryset(self, **hints): 

89 return self.field.remote_field.model.objects_inheritance.db_manager( 

90 hints=hints 

91 ).select_subclasses() 

92 

93 

94class InheritanceForeignKey(models.ForeignKey): 

95 forward_related_accessor_class = InheritanceForwardManyToOneDescriptor 

96 

97 

98# Uri model 

99# We use a custom UriManager, so we can override the queryset `get` 

100# method. This way we can normalize the uri field. 

101 

102 

103class UriQuerySet(models.query.QuerySet): 

104 def get(self, *args, **kwargs): 

105 if "uri" in kwargs: 

106 kwargs["uri"] = clean_uri(kwargs["uri"]) 

107 return super().get(*args, **kwargs) 

108 

109 

110class UriManager(models.Manager): 

111 def get_queryset(self): 

112 return UriQuerySet(self.model) 

113 

114 

115class Uri(GenericModel, models.Model): 

116 uri = models.URLField(blank=True, null=True, unique=True, max_length=255) 

117 domain = models.CharField(max_length=255, blank=True) 

118 rdf_link = models.URLField(blank=True) 

119 root_object = InheritanceForeignKey( 

120 RootObject, blank=True, null=True, on_delete=models.CASCADE 

121 ) 

122 # loaded set to True when RDF was loaded and parsed into the data model 

123 loaded = models.BooleanField(default=False) 

124 # Timestamp when file was loaded and parsed 

125 loaded_time = models.DateTimeField(blank=True, null=True) 

126 

127 objects = UriManager() 

128 

129 def __str__(self): 

130 return str(self.uri) 

131 

132 def get_web_object(self): 

133 result = { 

134 "relation_pk": self.pk, 

135 "relation_type": "uri", 

136 "related_root_object": self.root_object.name, 

137 "related_root_object_url": self.root_object.get_absolute_url(), 

138 "related_root_object_class_name": self.root_object.__class__.__name__.lower(), 

139 "uri": self.uri, 

140 } 

141 return result 

142 

143 def save(self, *args, **kwargs): 

144 self.clean() 

145 return super().save(*args, **kwargs) 

146 

147 def clean(self): 

148 self.uri = clean_uri(self.uri) 

149 if self.uri and not hasattr(self, "root_object"): 

150 try: 

151 definition, attributes = rdf.get_definition_and_attributes_from_uri( 

152 self.uri 

153 ) 

154 if definition.getattr("model", False) and attributes: 

155 app_label, model = definition.getattr("model").split(".", 1) 

156 ct = ContentType.objects.get_by_natural_key(app_label, model) 

157 obj = ct.model_class()(**attributes) 

158 obj.save() 

159 self.root_object = obj 

160 else: 

161 raise ImproperlyConfigured( 

162 f"{self.uri}: did not find matching rdf defintion" 

163 ) 

164 except Exception as e: 

165 raise ValidationError(f"{e}: {self.uri}")