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

1from urllib.parse import urlsplit 

2 

3from AcdhArcheAssets.uri_norm_rules import get_normalized_uri 

4from django.contrib.contenttypes.fields import GenericForeignKey 

5from django.contrib.contenttypes.models import ContentType 

6from django.core.exceptions import ImproperlyConfigured, ValidationError 

7from django.db import models 

8from model_utils.managers import InheritanceManager 

9 

10from apis_core.generic.abc import GenericModel 

11from apis_core.utils import rdf 

12from apis_core.utils import settings as apis_settings 

13 

14 

15class RootObject(GenericModel, models.Model): 

16 """ 

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

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

19 """ 

20 

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

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

23 # where the subject is a Person) 

24 self_contenttype = models.ForeignKey( 

25 ContentType, 

26 on_delete=models.deletion.CASCADE, 

27 null=True, 

28 blank=True, 

29 editable=False, 

30 ) 

31 objects = models.Manager() 

32 objects_inheritance = InheritanceManager() 

33 

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

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

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

37 

38 

39# Uri model 

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

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

42 

43 

44class UriQuerySet(models.query.QuerySet): 

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

46 if "uri" in kwargs: 

47 kwargs["uri"] = get_normalized_uri(kwargs["uri"]) 

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

49 

50 

51class UriManager(models.Manager): 

52 def get_queryset(self): 

53 return UriQuerySet(self.model) 

54 

55 

56class Uri(GenericModel, models.Model): 

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

58 content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE, null=True) 

59 object_id = models.PositiveIntegerField(null=True) 

60 content_object = GenericForeignKey() 

61 

62 objects = UriManager() 

63 

64 def __str__(self): 

65 return str(self.uri) 

66 

67 def get_web_object(self): 

68 result = { 

69 "relation_pk": self.pk, 

70 "relation_type": "uri", 

71 "related_root_object": self.content_object.name, 

72 "related_root_object_url": self.content_object.get_absolute_url(), 

73 "related_root_object_class_name": self.content_object.__class__.__name__.lower(), 

74 "uri": self.uri, 

75 } 

76 return result 

77 

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

79 self.clean() 

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

81 

82 def clean(self): 

83 self.uri = get_normalized_uri(self.uri) 

84 if self.uri and not hasattr(self, "content_object"): 

85 try: 

86 definition, attributes = rdf.get_definition_and_attributes_from_uri( 

87 self.uri 

88 ) 

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

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

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

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

93 obj.save() 

94 self.content_type = ContentType.objects.get_for_model(obj) 

95 self.object_id = obj.id 

96 else: 

97 raise ImproperlyConfigured( 

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

99 ) 

100 except Exception as e: 

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

102 

103 def internal(self) -> bool: 

104 my_netloc = urlsplit(self.uri).netloc 

105 return any( 

106 [my_netloc == urlsplit(uri).netloc for uri in apis_settings.internal_uris()] 

107 )