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, current_app, jsonify, request 

8from flask_babel import gettext 

9from flask_login import login_required 

10from heritrace.extensions import (get_custom_filter, get_display_rules, 

11 get_sparql) 

12from heritrace.utils.display_rules_utils import get_highest_priority_class 

13from heritrace.utils.shacl_utils import determine_shape_for_classes 

14from heritrace.utils.sparql_utils import get_entity_types 

15from heritrace.utils.virtuoso_utils import (VIRTUOSO_EXCLUDED_GRAPHS, 

16 is_virtuoso) 

17from SPARQLWrapper import JSON 

18 

19linked_resources_bp = Blueprint("linked_resources", __name__, url_prefix="/api/linked-resources") 

20 

21 

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

23 """ 

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

25 

26 An intermediate entity for virtual properties is identified by: 

27 1. Its type matches a virtual property target class 

28 

29 Args: 

30 entity_types: List of entity types for the referencing entity 

31 

32 Returns: 

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

34 """ 

35 if not entity_types: 

36 return False 

37 

38 # Get all display rules to check for virtual properties 

39 display_rules = get_display_rules() 

40 if not display_rules: 

41 return False 

42 

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

44 for entity_type in entity_types: 

45 for rule in display_rules: 

46 for prop_config in rule.get('displayProperties', []): 

47 if prop_config.get('isVirtual'): 

48 implementation = prop_config.get('implementedVia', {}) 

49 target_config = implementation.get('target', {}) 

50 

51 if target_config.get('class') == entity_type: 

52 current_app.logger.debug(f"Found virtual property target entity type: {entity_type}") 

53 return True 

54 

55 return False 

56 

57 

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

59 """ 

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

61  

62 Args: 

63 entity_types: List of entity types 

64  

65 Returns: 

66 Tuple of (is_proxy, connecting_predicate) 

67 """ 

68 if not entity_types: 

69 return False, "" 

70 

71 display_rules = get_display_rules() 

72 if not display_rules: 

73 return False, "" 

74 

75 for entity_type in entity_types: 

76 for rule in display_rules: 

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

78 intermediate_config = prop.get("intermediateRelation") 

79 if intermediate_config and intermediate_config.get("class") == entity_type: 

80 return True, prop["property"] 

81 

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

83 nested_intermediate_config = display_rule.get("intermediateRelation") 

84 if nested_intermediate_config and nested_intermediate_config.get("class") == entity_type: 

85 return True, prop["property"] 

86 

87 return False, "" 

88 

89 

90def _resolve_proxy_entity(subject_uri: str, predicate: str, connecting_predicate: str) -> tuple[str, str]: 

91 """ 

92 Resolve proxy entities to their source entities. 

93  

94 Args: 

95 subject_uri: URI of the proxy entity 

96 predicate: Original predicate 

97 connecting_predicate: The predicate that connects source to proxy 

98  

99 Returns: 

100 Tuple of (final_subject_uri, final_predicate) 

101 """ 

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

103 sparql = get_sparql() 

104 proxy_query_parts = [ 

105 "SELECT DISTINCT ?source WHERE {", 

106 ] 

107 

108 if is_virtuoso: 

109 proxy_query_parts.extend([ 

110 " GRAPH ?g {", 

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

112 " }", 

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

114 ]) 

115 else: 

116 proxy_query_parts.extend([ 

117 f" ?source <{connecting_predicate}> <{subject_uri}> ." 

118 ]) 

119 

120 proxy_query_parts.append("} LIMIT 1") 

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

122 

123 try: 

124 sparql.setQuery(proxy_query) 

125 sparql.setReturnFormat(JSON) 

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

127 

128 proxy_bindings = proxy_results.get("results", {}).get("bindings", []) 

129 if proxy_bindings: 

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

131 

132 return source_uri, connecting_predicate 

133 

134 except Exception as e: 

135 current_app.logger.error(f"Error resolving proxy entity {subject_uri}: {e}") 

136 

137 return subject_uri, predicate 

138 

139def get_paginated_inverse_references(subject_uri: str, limit: int, offset: int) -> tuple[list[dict], bool]: 

140 """ 

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

142 

143 Args: 

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

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

146 offset: Number of references to skip. 

147 

148 Returns: 

149 A tuple containing: 

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

151 - Boolean indicating if there are more references. 

152 """ 

153 sparql = get_sparql() 

154 custom_filter = get_custom_filter() 

155 references = [] 

156 query_limit = limit + 1 

157 

158 try: 

159 query_parts = [ 

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

161 ] 

162 if is_virtuoso: 

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

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

165 else: 

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

167 

168 query_parts.extend([ 

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

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

171 f"}} ORDER BY ?s OFFSET {offset} LIMIT {query_limit}" # Use query_limit 

172 ]) 

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

174 

175 sparql.setQuery(main_query) 

176 sparql.setReturnFormat(JSON) 

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

178 

179 bindings = results.get("results", {}).get("bindings", []) 

180 

181 # Determine if there are more results 

182 has_more = len(bindings) > limit 

183 

184 # Process only up to 'limit' results 

185 results_to_process = bindings[:limit] 

186 

187 for result in results_to_process: 

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

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

190 

191 types = get_entity_types(subject) 

192 

193 if _is_virtual_property_intermediate_entity(types): 

194 continue 

195 

196 highest_priority_type = get_highest_priority_class(types) 

197 shape = determine_shape_for_classes(types) 

198 

199 is_proxy, connecting_predicate = _is_proxy_entity(types) 

200 if is_proxy: 

201 final_subject, final_predicate = _resolve_proxy_entity(subject, predicate, connecting_predicate) 

202 else: 

203 final_subject, final_predicate = subject, predicate 

204 

205 if final_subject != subject: 

206 final_types = get_entity_types(final_subject) 

207 final_highest_priority_type = get_highest_priority_class(final_types) 

208 final_shape = determine_shape_for_classes(final_types) 

209 else: 

210 final_types = types 

211 final_highest_priority_type = highest_priority_type 

212 final_shape = shape 

213 

214 label = custom_filter.human_readable_entity(final_subject, (final_highest_priority_type, final_shape)) 

215 type_label = custom_filter.human_readable_class((final_highest_priority_type, final_shape)) if final_highest_priority_type else None 

216 

217 references.append({ 

218 "subject": final_subject, 

219 "predicate": final_predicate, 

220 "predicate_label": custom_filter.human_readable_predicate(final_predicate, (final_highest_priority_type, final_shape)), 

221 "type_label": type_label, 

222 "label": label 

223 }) 

224 

225 return references, has_more 

226 

227 except Exception as e: 

228 tb_str = traceback.format_exc() 

229 current_app.logger.error(f"Error fetching inverse references for {subject_uri}: {e}\n{tb_str}") 

230 return [], False 

231 

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

233@login_required 

234def get_linked_resources_api(): 

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

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

237 try: 

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

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

240 except ValueError: 

241 return jsonify({"status": "error", "message": gettext("Invalid limit or offset parameter")}), 400 

242 

243 if not subject_uri: 

244 return jsonify({"status": "error", "message": gettext("Missing subject_uri parameter")}), 400 

245 

246 if limit <= 0 or offset < 0: 

247 return jsonify({"status": "error", "message": gettext("Limit must be positive and offset non-negative")}), 400 

248 

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

250 

251 return jsonify({ 

252 "status": "success", 

253 "results": references, 

254 "has_more": has_more 

255 })