Coverage for apis_core/generic/serializers.py: 41%

1import re 

2 

3from AcdhArcheAssets.uri_norm_rules import get_rules 

4from django.contrib.contenttypes.models import ContentType 

5from django.shortcuts import get_object_or_404 

6from rdflib import Graph, Literal, Namespace, URIRef 

7from rdflib.namespace import OWL, RDF, RDFS 

8from rest_framework.reverse import reverse 

9from rest_framework.serializers import ( 

10 BaseSerializer, 

11 CharField, 

12 HyperlinkedModelSerializer, 

13 HyperlinkedRelatedField, 

14 IntegerField, 

15 Serializer, 

16 SerializerMethodField, 

17) 

18 

19from apis_core.generic.utils.rdf_namespace import APPELLATION, ATTRIBUTES, CRM 

20from apis_core.utils.settings import rdf_namespace_prefix 

21 

22 

23class GenericHyperlinkedRelatedField(HyperlinkedRelatedField): 

24 def get_url(self, obj, view_name, request, format): 

25 contenttype = ContentType.objects.get_for_model(obj, for_concrete_model=True) 

26 url_kwargs = {"contenttype": contenttype, "pk": obj.pk} 

27 return reverse( 

28 "apis_core:generic:genericmodelapi-detail", 

29 kwargs=url_kwargs, 

30 request=request, 

31 format=format, 

32 ) 

33 

34 def use_pk_only_optimization(self): 

35 # We have the complete object instance already. We don't need 

36 # to run the 'only get the pk for this relationship' code. 

37 return False 

38 

39 

40class GenericHyperlinkedIdentityField(GenericHyperlinkedRelatedField): 

41 def __init__(self, view_name=None, **kwargs): 

42 assert view_name is not None, "The `view_name` argument is required." 

43 kwargs["read_only"] = True 

44 kwargs["source"] = "*" 

45 super().__init__(view_name, **kwargs) 

46 

47 

48class GenericHyperlinkedModelSerializer(HyperlinkedModelSerializer): 

49 serializer_related_field = GenericHyperlinkedRelatedField 

50 serializer_url_field = GenericHyperlinkedIdentityField 

51 

52 

53def serializer_factory( 

54 model, 

55 serializer=GenericHyperlinkedModelSerializer, 

56 action="detail", 

57 fields="__all__", 

58 **kwargs, 

59): 

60 defaultmeta = type(str("Meta"), (object,), {"fields": fields}) 

61 meta = getattr(serializer, "Meta", defaultmeta) 

62 meta.model = model 

63 serializer = type( 

64 str("%s%sModelSerializer" % (model._meta.object_name, action)), 

65 (serializer,), 

66 {"Meta": meta}, 

67 ) 

68 return serializer 

69 

70 

71class ContentTypeInstanceSerializer(Serializer): 

72 id = IntegerField(required=True) 

73 content_type = CharField(required=True) 

74 

75 def to_internal_value(self, data): 

76 data = super().to_internal_value(data) 

77 app_label, model = data.get("content_type").split(".") 

78 content_type = get_object_or_404(ContentType, app_label=app_label, model=model) 

79 return get_object_or_404(content_type.model_class(), pk=data.get("id")) 

80 

81 

82class SimpleObjectSerializer(Serializer): 

83 url = GenericHyperlinkedIdentityField( 

84 view_name="apis_core:generic:genericmodelapi-detail" 

85 ) 

86 label = SerializerMethodField() 

87 content_type_key = SerializerMethodField() 

88 content_type_name = SerializerMethodField() 

89 

90 class Meta: 

91 fields = ["url", "label", "content_type_key", "content_type_name"] 

92 

93 def get_label(self, obj) -> str: 

94 return str(obj) 

95 

96 def get_content_type_key(self, obj) -> str: 

97 content_type = ContentType.objects.get_for_model(obj) 

98 return f"{content_type.app_label}.{content_type.model}" 

99 

100 def get_content_type_name(self, obj) -> str: 

101 content_type = ContentType.objects.get_for_model(obj) 

102 return content_type.name 

103 

104 

105class GenericModelCidocSerializer(BaseSerializer): 

106 def __init__(self, *args, **kwargs): 

107 self.rdf_nsp_base = rdf_namespace_prefix() 

108 self.appellation_nsp_prefix = f"{self.rdf_nsp_base}-appellation" 

109 self.attr_nsp_prefix = f"{self.rdf_nsp_base}-attr" 

110 super().__init__(*args, **kwargs) 

111 

112 def create_sameas(self, g, instance): 

113 # add the ID as APIS Identifier 

114 apis_id = URIRef(ATTRIBUTES[f"apis-identifier_{instance.pk}"]) 

115 g.add((apis_id, RDF.type, CRM.E42_Identifier)) 

116 g.add((apis_id, RDFS.label, Literal(instance.pk))) 

117 

118 # APIS internal identifier type 

119 apis_id_type = URIRef(ATTRIBUTES["apis-identifier_type"]) 

120 g.add((apis_id, CRM.P2_has_type, apis_id_type)) 

121 g.add((apis_id_type, RDF.type, CRM.E55_Type)) 

122 g.add((apis_id_type, RDFS.label, Literal("APIS internal identifier"))) 

123 g.add((self.instance_uri, CRM.P1_is_identified_by, apis_id)) 

124 

125 for uri in instance.uri_set(): 

126 g.add((self.instance_uri, OWL.sameAs, URIRef(uri.uri))) 

127 

128 for x in get_rules(): 

129 if m := re.match(x["match"], uri.uri): 

130 id_type = URIRef(ATTRIBUTES[x["name"] + "-identifier_type"]) 

131 g.add((id_type, RDF.type, CRM.E55_Type)) 

132 g.add((id_type, RDFS.label, Literal(x["name"] + " ID"))) 

133 id_uri = URIRef( 

134 ATTRIBUTES[x["name"] + f"-identifier_{instance.pk}"] 

135 ) 

136 g.add((id_uri, RDF.type, CRM.E42_Identifier)) 

137 g.add((id_uri, CRM.P2_has_type, id_type)) 

138 g.add( 

139 ( 

140 self.instance_uri, 

141 CRM.P1_is_identified_by, 

142 id_uri, 

143 ) 

144 ) 

145 g.add((id_uri, RDFS.label, Literal(m[1]))) 

146 

147 return g 

148 

149 def to_representation(self, instance): 

150 g = Graph() 

151 

152 g.namespace_manager.bind("crm", CRM, replace=True) 

153 g.namespace_manager.bind("owl", OWL, replace=True) 

154 

155 g.namespace_manager.bind(self.appellation_nsp_prefix, APPELLATION, replace=True) 

156 g.namespace_manager.bind(self.attr_nsp_prefix, ATTRIBUTES, replace=True) 

157 

158 self.instance_namespace = Namespace(instance.get_namespace_uri()) 

159 g.namespace_manager.bind( 

160 instance.get_namespace_prefix(), self.instance_namespace 

161 ) 

162 

163 self.instance_uri = URIRef(self.instance_namespace[str(instance.id)]) 

164 

165 # Add properties 

166 self.appellation_uri = URIRef(APPELLATION[str(instance.id)]) 

167 g.add( 

168 ( 

169 self.appellation_uri, 

170 RDF.type, 

171 CRM.E33_E41_Linguistic_Appellation, 

172 ) 

173 ) 

174 g.add( 

175 ( 

176 self.instance_uri, 

177 CRM.P1_is_identified_by, 

178 self.appellation_uri, 

179 ) 

180 ) 

181 g.add((self.appellation_uri, RDFS.label, Literal(str(instance)))) 

182 g.add((self.instance_uri, RDFS.label, Literal(str(instance)))) 

183 # Add RDF types 

184 for rdf_type in instance.get_rdf_types(): 

185 g.add((self.instance_uri, RDF.type, rdf_type)) 

186 # Add sameAs links 

187 g = self.create_sameas(g, instance) 

188 

189 return g