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

59 statements  

« prev     ^ index     » next       coverage.py v7.6.8, created at 2024-12-20 09:24 +0000

1from urllib.parse import urlsplit 

2 

3from django.contrib.contenttypes.models import ContentType 

4from django.core.exceptions import ImproperlyConfigured, ValidationError 

5from django.db import models 

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

7from model_utils.managers import InheritanceManager 

8 

9from apis_core.generic.abc import GenericModel 

10from apis_core.utils import rdf 

11from apis_core.utils import settings as apis_settings 

12from apis_core.utils.normalize import clean_uri 

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 

39class InheritanceForwardManyToOneDescriptor(ForwardManyToOneDescriptor): 

40 def get_queryset(self, **hints): 

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

42 hints=hints 

43 ).select_subclasses() 

44 

45 

46class InheritanceForeignKey(models.ForeignKey): 

47 forward_related_accessor_class = InheritanceForwardManyToOneDescriptor 

48 

49 

50# Uri model 

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

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

53 

54 

55class UriQuerySet(models.query.QuerySet): 

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

57 if "uri" in kwargs: 

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

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

60 

61 

62class UriManager(models.Manager): 

63 def get_queryset(self): 

64 return UriQuerySet(self.model) 

65 

66 

67class Uri(GenericModel, models.Model): 

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

69 root_object = InheritanceForeignKey( 

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

71 ) 

72 

73 objects = UriManager() 

74 

75 def __str__(self): 

76 return str(self.uri) 

77 

78 def get_web_object(self): 

79 result = { 

80 "relation_pk": self.pk, 

81 "relation_type": "uri", 

82 "related_root_object": self.root_object.name, 

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

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

85 "uri": self.uri, 

86 } 

87 return result 

88 

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

90 self.clean() 

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

92 

93 def clean(self): 

94 self.uri = clean_uri(self.uri) 

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

96 try: 

97 definition, attributes = rdf.get_definition_and_attributes_from_uri( 

98 self.uri 

99 ) 

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

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

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

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

104 obj.save() 

105 self.root_object = obj 

106 else: 

107 raise ImproperlyConfigured( 

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

109 ) 

110 except Exception as e: 

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

112 

113 def internal(self) -> bool: 

114 my_netloc = urlsplit(self.uri).netloc 

115 return any( 

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

117 )