Coverage for apis_core/uris/models.py: 65%

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 

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# Uri model 

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

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

17 

18 

19class UriQuerySet(models.query.QuerySet): 

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

21 if "uri" in kwargs: 

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

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

24 

25 

26class UriManager(models.Manager): 

27 def get_queryset(self): 

28 return UriQuerySet(self.model) 

29 

30 

31class Uri(GenericModel, models.Model): 

32 """ 

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

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

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

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

37 the external object). 

38 """ 

39 

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

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

42 object_id = models.PositiveIntegerField(null=True) 

43 content_object = GenericForeignKey() 

44 

45 objects = UriManager() 

46 

47 def __str__(self): 

48 return str(self.uri) 

49 

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

51 self.clean() 

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

53 

54 def clean(self): 

55 self.uri = get_normalized_uri(self.uri) 

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

57 try: 

58 result = rdf.get_something_from_uri(self.uri) 

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

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

61 obj.save() 

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

63 self.object_id = obj.id 

64 else: 

65 raise ImproperlyConfigured( 

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

67 ) 

68 except Exception as e: 

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

70 

71 def internal(self) -> bool: 

72 my_netloc = urlsplit(self.uri).netloc 

73 return any( 

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

75 ) 

76 

77 def short_label(self) -> str: 

78 if self.internal(): 

79 return "local" 

80 for x in get_rules(): 

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

82 return x["name"] 

83 return urlsplit(self.uri).netloc