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

1import re 

2from urllib.parse import urlsplit 

3 

4from AcdhArcheAssets.uri_norm_rules import get_normalized_uri, get_rules 

5from django.contrib.contenttypes.fields import GenericForeignKey 

6from django.contrib.contenttypes.models import ContentType 

7from django.core.exceptions import ImproperlyConfigured, ValidationError 

8from django.db import models 

9from model_utils.managers import InheritanceManager 

10 

11from apis_core.generic.abc import GenericModel 

12from apis_core.utils import rdf 

13from apis_core.utils import settings as apis_settings 

14 

15 

16class RootObject(GenericModel, models.Model): 

17 """ 

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

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

20 """ 

21 

22 objects = models.Manager() 

23 objects_inheritance = InheritanceManager() 

24 

25 

26# Uri model 

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

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

29 

30 

31class UriQuerySet(models.query.QuerySet): 

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

33 if "uri" in kwargs: 

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

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

36 

37 

38class UriManager(models.Manager): 

39 def get_queryset(self): 

40 return UriQuerySet(self.model) 

41 

42 

43class Uri(GenericModel, models.Model): 

44 """ 

45 The URI model provides a way of storing globally scoped identifiers for 

46 objects. For the objects in our application, we store both internal URIs 

47 (URIs that we created) as well as external URIs (when we want to link the 

48 object to an external resource, asserting that this object is the "same as" 

49 the external object). 

50 """ 

51 

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

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

54 object_id = models.PositiveIntegerField(null=True) 

55 content_object = GenericForeignKey() 

56 

57 objects = UriManager() 

58 

59 def __str__(self): 

60 return str(self.uri) 

61 

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

63 self.clean() 

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

65 

66 def clean(self): 

67 self.uri = get_normalized_uri(self.uri) 

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

69 try: 

70 result = rdf.get_something_from_uri(self.uri) 

71 if model := result.getattr("model", False): 

72 obj = model(**result.get("attributes", {})) 

73 obj.save() 

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

75 self.object_id = obj.id 

76 else: 

77 raise ImproperlyConfigured( 

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

79 ) 

80 except Exception as e: 

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

82 

83 def internal(self) -> bool: 

84 my_netloc = urlsplit(self.uri).netloc 

85 return any( 

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

87 ) 

88 

89 def short_label(self) -> str: 

90 if self.internal(): 

91 return "local" 

92 for x in get_rules(): 

93 if re.match(x["match"], self.uri): 

94 return x["name"] 

95 return urlsplit(self.uri).netloc