Coverage for heritrace/routes/linked_resources.py: 97%

1# SPDX-FileCopyrightText: 2025 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from flask import Blueprint, Response, current_app, jsonify, request 

6from flask_babel import gettext 

7from flask_login import login_required 

8from SPARQLWrapper import JSON 

9from SPARQLWrapper.SPARQLExceptions import SPARQLWrapperException 

10 

11from heritrace.extensions import ( 

12 get_custom_filter, 

13 get_display_rules, 

14 get_sparql, 

15 get_sparql_bindings, 

16) 

17from heritrace.utils.display_rules_utils import get_highest_priority_class 

18from heritrace.utils.shacl_utils import determine_shape_for_classes 

19from heritrace.utils.sparql_utils import get_entity_types 

20from heritrace.utils.virtuoso_utils import VIRTUOSO_EXCLUDED_GRAPHS, is_virtuoso 

21 

22linked_resources_bp = Blueprint( 

23 "linked_resources", __name__, url_prefix="/api/linked-resources" 

24) 

25 

26 

27def _is_virtual_property_intermediate_entity(entity_types: list) -> bool: 

28 """ 

29 Check if an entity is an intermediate entity created for virtual properties. 

30 

31 An intermediate entity for virtual properties is identified by: 

32 1. Its type matches a virtual property target class 

33 

34 Args: 

35 entity_types: List of entity types for the referencing entity 

36 

37 Returns: 

38 True if this is a virtual property intermediate entity, False otherwise 

39 """ 

40 if not entity_types: 

41 return False 

42 

43 # Get all display rules to check for virtual properties 

44 display_rules = get_display_rules() 

45 if not display_rules: 

46 return False 

47 

48 # Check if any of the entity types matches a virtual property target class 

49 for entity_type in entity_types: 

50 for rule in display_rules: 

51 for prop_config in rule.get("displayProperties", []): 

52 if prop_config.get("isVirtual"): 

53 implementation = prop_config.get("implementedVia", {}) 

54 target_config = implementation.get("target", {}) 

55 

56 if target_config.get("class") == entity_type: 

57 current_app.logger.debug( 

58 "Found virtual property target entity type: %s", entity_type 

59 ) 

60 return True 

61 

62 return False 

63 

64 

65def _is_proxy_entity(entity_types: list) -> tuple[bool, str]: 

66 """ 

67 Check if an entity is a proxy (intermediate relation) entity based on display rules. 

68 

69 Args: 

70 entity_types: List of entity types 

71 

72 Returns: 

73 Tuple of (is_proxy, connecting_predicate) 

74 """ 

75 if not entity_types: 

76 return False, "" 

77 

78 display_rules = get_display_rules() 

79 if not display_rules: 

80 return False, "" 

81 

82 for entity_type in entity_types: 

83 for rule in display_rules: 

84 for prop in rule.get("displayProperties", []): 

85 intermediate_config = prop.get("intermediateRelation") 

86 if ( 

87 intermediate_config 

88 and intermediate_config.get("class") == entity_type 

89 ): 

90 return True, prop["property"] 

91 

92 for display_rule in prop.get("displayRules", []): 

93 nested_intermediate_config = display_rule.get( 

94 "intermediateRelation" 

95 ) 

96 if ( 

97 nested_intermediate_config 

98 and nested_intermediate_config.get("class") == entity_type 

99 ): 

100 return True, prop["property"] 

101 

102 return False, "" 

103 

104 

105def _resolve_proxy_entity( 

106 subject_uri: str, predicate: str, connecting_predicate: str 

107) -> tuple[str, str]: 

108 """ 

109 Resolve proxy entities to their source entities. 

110 

111 Args: 

112 subject_uri: URI of the proxy entity 

113 predicate: Original predicate 

114 connecting_predicate: The predicate that connects source to proxy 

115 

116 Returns: 

117 Tuple of (final_subject_uri, final_predicate) 

118 """ 

119 # Query to find the source entity that points to this proxy 

120 sparql = get_sparql() 

121 proxy_query_parts = [ 

122 "SELECT DISTINCT ?source WHERE {", 

123 ] 

124 

125 if is_virtuoso(): 

126 proxy_query_parts.extend( 

127 [ 

128 " GRAPH ?g {", 

129 f" ?source <{connecting_predicate}> <{subject_uri}> .", 

130 " }", 

131 f" FILTER(?g NOT IN (<{'>, <'.join(VIRTUOSO_EXCLUDED_GRAPHS)}>))", 

132 ] 

133 ) 

134 else: 

135 proxy_query_parts.extend( 

136 [f" ?source <{connecting_predicate}> <{subject_uri}> ."] 

137 ) 

138 

139 proxy_query_parts.append("} LIMIT 1") 

140 proxy_query = "\n".join(proxy_query_parts) 

141 

142 sparql.setQuery(proxy_query) 

143 sparql.setReturnFormat(JSON) 

144 proxy_results = sparql.query().convert() 

145 

146 proxy_bindings = get_sparql_bindings(proxy_results) 

147 if proxy_bindings: 

148 source_uri = proxy_bindings[0]["source"]["value"] 

149 

150 return source_uri, connecting_predicate 

151 

152 return subject_uri, predicate 

153 

154 

155def get_paginated_inverse_references( 

156 subject_uri: str, limit: int, offset: int 

157) -> tuple[list[dict], bool]: 

158 """ 

159 Get paginated entities that reference this entity using the limit+1 strategy. 

160 

161 Args: 

162 subject_uri: URI of the entity to find references to. 

163 limit: Maximum number of references to return per page. 

164 offset: Number of references to skip. 

165 

166 Returns: 

167 A tuple containing: 

168 - List of dictionaries containing reference information (max 'limit' items). 

169 - Boolean indicating if there are more references. 

170 """ 

171 sparql = get_sparql() 

172 custom_filter = get_custom_filter() 

173 references = [] 

174 query_limit = limit + 1 

175 

176 query_parts = [ 

177 "SELECT DISTINCT ?s ?p WHERE {", 

178 ] 

179 if is_virtuoso(): 

180 query_parts.append(" GRAPH ?g { ?s ?p ?o . }") 

181 query_parts.append( 

182 f" FILTER(?g NOT IN (<{'>, <'.join(VIRTUOSO_EXCLUDED_GRAPHS)}>))" 

183 ) 

184 query_parts.append(f" FILTER(?o = <{subject_uri}>)") 

185 else: 

186 query_parts.append(f" ?s ?p <{subject_uri}> .") 

187 

188 query_parts.extend( 

189 [ 

190 " FILTER(?p != <http://www.w3.org/1999/02/22-rdf-syntax-ns#type>)", 

191 f"}} ORDER BY ?s OFFSET {offset} LIMIT {query_limit}", 

192 ] 

193 ) 

194 main_query = "\n".join(query_parts) 

195 

196 sparql.setQuery(main_query) 

197 sparql.setReturnFormat(JSON) 

198 results = sparql.query().convert() 

199 

200 bindings = get_sparql_bindings(results) 

201 

202 # Determine if there are more results 

203 has_more = len(bindings) > limit 

204 

205 # Process only up to 'limit' results 

206 results_to_process = bindings[:limit] 

207 

208 for result in results_to_process: 

209 subject = result["s"]["value"] 

210 predicate = result["p"]["value"] 

211 

212 types = get_entity_types(subject) 

213 

214 if _is_virtual_property_intermediate_entity(types): 

215 continue 

216 

217 highest_priority_type = get_highest_priority_class(types) 

218 shape = determine_shape_for_classes(types) 

219 

220 is_proxy, connecting_predicate = _is_proxy_entity(types) 

221 if is_proxy: 

222 final_subject, final_predicate = _resolve_proxy_entity( 

223 subject, predicate, connecting_predicate 

224 ) 

225 else: 

226 final_subject, final_predicate = subject, predicate 

227 

228 if final_subject != subject: 

229 final_types = get_entity_types(final_subject) 

230 final_highest_priority_type = get_highest_priority_class(final_types) 

231 final_shape = determine_shape_for_classes(final_types) 

232 else: 

233 final_types = types 

234 final_highest_priority_type = highest_priority_type 

235 final_shape = shape 

236 

237 label = custom_filter.human_readable_entity( 

238 final_subject, (final_highest_priority_type, final_shape) 

239 ) 

240 type_label = ( 

241 custom_filter.human_readable_class( 

242 (final_highest_priority_type, final_shape) 

243 ) 

244 if final_highest_priority_type 

245 else None 

246 ) 

247 

248 references.append( 

249 { 

250 "subject": final_subject, 

251 "predicate": final_predicate, 

252 "predicate_label": custom_filter.human_readable_predicate( 

253 final_predicate, (final_highest_priority_type, final_shape) 

254 ), 

255 "type_label": type_label, 

256 "label": label, 

257 } 

258 ) 

259 

260 return references, has_more 

261 

262 

263@linked_resources_bp.route("/", methods=["GET"]) 

264@login_required 

265def get_linked_resources_api() -> Response | tuple[Response, int]: 

266 """API endpoint to fetch paginated linked resources (inverse references).""" 

267 subject_uri = request.args.get("subject_uri") 

268 try: 

269 limit = int(request.args.get("limit", 5)) 

270 offset = int(request.args.get("offset", 0)) 

271 except ValueError: 

272 return jsonify( 

273 {"status": "error", "message": gettext("Invalid limit or offset parameter")} 

274 ), 400 

275 

276 if not subject_uri: 

277 return jsonify( 

278 {"status": "error", "message": gettext("Missing subject_uri parameter")} 

279 ), 400 

280 

281 if limit <= 0 or offset < 0: 

282 return jsonify( 

283 { 

284 "status": "error", 

285 "message": gettext("Limit must be positive and offset non-negative"), 

286 } 

287 ), 400 

288 

289 try: 

290 references, has_more = get_paginated_inverse_references( 

291 subject_uri, limit, offset 

292 ) 

293 except (SPARQLWrapperException, OSError): 

294 current_app.logger.exception( 

295 "Error fetching inverse references for %s", subject_uri 

296 ) 

297 return jsonify( 

298 {"status": "error", "message": gettext("Error loading linked resources.")} 

299 ), 502 

300 

301 return jsonify({"status": "success", "results": references, "has_more": has_more})