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 

5import traceback 

6 

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

8from flask_babel import gettext 

9from flask_login import login_required 

10from SPARQLWrapper import JSON 

11from SPARQLWrapper.SPARQLExceptions import SPARQLWrapperException 

12 

13from heritrace.extensions import ( 

14 get_custom_filter, 

15 get_display_rules, 

16 get_sparql, 

17 get_sparql_bindings, 

18) 

19from heritrace.utils.display_rules_utils import get_highest_priority_class 

20from heritrace.utils.shacl_utils import determine_shape_for_classes 

21from heritrace.utils.sparql_utils import get_entity_types 

22from heritrace.utils.virtuoso_utils import VIRTUOSO_EXCLUDED_GRAPHS, is_virtuoso 

23 

24linked_resources_bp = Blueprint( 

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

26) 

27 

28 

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

30 """ 

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

32 

33 An intermediate entity for virtual properties is identified by: 

34 1. Its type matches a virtual property target class 

35 

36 Args: 

37 entity_types: List of entity types for the referencing entity 

38 

39 Returns: 

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

41 """ 

42 if not entity_types: 

43 return False 

44 

45 # Get all display rules to check for virtual properties 

46 display_rules = get_display_rules() 

47 if not display_rules: 

48 return False 

49 

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

51 for entity_type in entity_types: 

52 for rule in display_rules: 

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

54 if prop_config.get("isVirtual"): 

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

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

57 

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

59 current_app.logger.debug( 

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

61 ) 

62 return True 

63 

64 return False 

65 

66 

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

68 """ 

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

70 

71 Args: 

72 entity_types: List of entity types 

73 

74 Returns: 

75 Tuple of (is_proxy, connecting_predicate) 

76 """ 

77 if not entity_types: 

78 return False, "" 

79 

80 display_rules = get_display_rules() 

81 if not display_rules: 

82 return False, "" 

83 

84 for entity_type in entity_types: 

85 for rule in display_rules: 

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

87 intermediate_config = prop.get("intermediateRelation") 

88 if ( 

89 intermediate_config 

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

91 ): 

92 return True, prop["property"] 

93 

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

95 nested_intermediate_config = display_rule.get( 

96 "intermediateRelation" 

97 ) 

98 if ( 

99 nested_intermediate_config 

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

101 ): 

102 return True, prop["property"] 

103 

104 return False, "" 

105 

106 

107def _resolve_proxy_entity( 

108 subject_uri: str, predicate: str, connecting_predicate: str 

109) -> tuple[str, str]: 

110 """ 

111 Resolve proxy entities to their source entities. 

112 

113 Args: 

114 subject_uri: URI of the proxy entity 

115 predicate: Original predicate 

116 connecting_predicate: The predicate that connects source to proxy 

117 

118 Returns: 

119 Tuple of (final_subject_uri, final_predicate) 

120 """ 

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

122 sparql = get_sparql() 

123 proxy_query_parts = [ 

124 "SELECT DISTINCT ?source WHERE {", 

125 ] 

126 

127 if is_virtuoso: 

128 proxy_query_parts.extend( 

129 [ 

130 " GRAPH ?g {", 

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

132 " }", 

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

134 ] 

135 ) 

136 else: 

137 proxy_query_parts.extend( 

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

139 ) 

140 

141 proxy_query_parts.append("} LIMIT 1") 

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

143 

144 try: 

145 sparql.setQuery(proxy_query) 

146 sparql.setReturnFormat(JSON) 

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

148 

149 proxy_bindings = get_sparql_bindings(proxy_results) 

150 if proxy_bindings: 

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

152 

153 return source_uri, connecting_predicate 

154 

155 except SPARQLWrapperException: 

156 current_app.logger.exception("Error resolving proxy entity %s", subject_uri) 

157 

158 return subject_uri, predicate 

159 

160 

161def get_paginated_inverse_references( 

162 subject_uri: str, limit: int, offset: int 

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

164 """ 

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

166 

167 Args: 

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

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

170 offset: Number of references to skip. 

171 

172 Returns: 

173 A tuple containing: 

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

175 - Boolean indicating if there are more references. 

176 """ 

177 sparql = get_sparql() 

178 custom_filter = get_custom_filter() 

179 references = [] 

180 query_limit = limit + 1 

181 

182 try: 

183 query_parts = [ 

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

185 ] 

186 if is_virtuoso: 

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

188 query_parts.append( 

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

190 ) 

191 else: 

192 query_parts.append(" ?s ?p ?o .") 

193 

194 query_parts.extend( 

195 [ 

196 f" FILTER(?o = <{subject_uri}>)", 

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

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

199 ] 

200 ) 

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

202 

203 sparql.setQuery(main_query) 

204 sparql.setReturnFormat(JSON) 

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

206 

207 bindings = get_sparql_bindings(results) 

208 

209 # Determine if there are more results 

210 has_more = len(bindings) > limit 

211 

212 # Process only up to 'limit' results 

213 results_to_process = bindings[:limit] 

214 

215 for result in results_to_process: 

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

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

218 

219 types = get_entity_types(subject) 

220 

221 if _is_virtual_property_intermediate_entity(types): 

222 continue 

223 

224 highest_priority_type = get_highest_priority_class(types) 

225 shape = determine_shape_for_classes(types) 

226 

227 is_proxy, connecting_predicate = _is_proxy_entity(types) 

228 if is_proxy: 

229 final_subject, final_predicate = _resolve_proxy_entity( 

230 subject, predicate, connecting_predicate 

231 ) 

232 else: 

233 final_subject, final_predicate = subject, predicate 

234 

235 if final_subject != subject: 

236 final_types = get_entity_types(final_subject) 

237 final_highest_priority_type = get_highest_priority_class(final_types) 

238 final_shape = determine_shape_for_classes(final_types) 

239 else: 

240 final_types = types 

241 final_highest_priority_type = highest_priority_type 

242 final_shape = shape 

243 

244 label = custom_filter.human_readable_entity( 

245 final_subject, (final_highest_priority_type, final_shape) 

246 ) 

247 type_label = ( 

248 custom_filter.human_readable_class( 

249 (final_highest_priority_type, final_shape) 

250 ) 

251 if final_highest_priority_type 

252 else None 

253 ) 

254 

255 references.append( 

256 { 

257 "subject": final_subject, 

258 "predicate": final_predicate, 

259 "predicate_label": custom_filter.human_readable_predicate( 

260 final_predicate, (final_highest_priority_type, final_shape) 

261 ), 

262 "type_label": type_label, 

263 "label": label, 

264 } 

265 ) 

266 

267 except Exception: 

268 traceback.format_exc() 

269 current_app.logger.exception( 

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

271 ) 

272 return [], False 

273 else: 

274 return references, has_more 

275 

276 

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

278@login_required 

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

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

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

282 try: 

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

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

285 except ValueError: 

286 return jsonify( 

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

288 ), 400 

289 

290 if not subject_uri: 

291 return jsonify( 

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

293 ), 400 

294 

295 if limit <= 0 or offset < 0: 

296 return jsonify( 

297 { 

298 "status": "error", 

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

300 } 

301 ), 400 

302 

303 references, has_more = get_paginated_inverse_references(subject_uri, limit, offset) 

304 

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