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

1import traceback 

2 

3from flask import Blueprint, current_app, jsonify, request 

4from flask_babel import gettext 

5from flask_login import login_required 

6from heritrace.extensions import (get_custom_filter, get_display_rules, 

7 get_sparql) 

8from heritrace.utils.display_rules_utils import get_highest_priority_class 

9from heritrace.utils.shacl_utils import determine_shape_for_classes 

10from heritrace.utils.sparql_utils import get_entity_types 

11from heritrace.utils.virtuoso_utils import (VIRTUOSO_EXCLUDED_GRAPHS, 

12 is_virtuoso) 

13from SPARQLWrapper import JSON 

14 

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

16 

17 

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

19 """ 

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

21 

22 An intermediate entity for virtual properties is identified by: 

23 1. Its type matches a virtual property target class 

24 

25 Args: 

26 entity_types: List of entity types for the referencing entity 

27 

28 Returns: 

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

30 """ 

31 if not entity_types: 

32 return False 

33 

34 # Get all display rules to check for virtual properties 

35 display_rules = get_display_rules() 

36 if not display_rules: 

37 return False 

38 

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

40 for entity_type in entity_types: 

41 for rule in display_rules: 

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

43 if prop_config.get('isVirtual'): 

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

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

46 

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

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

49 return True 

50 

51 return False 

52 

53 

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

55 """ 

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

57  

58 Args: 

59 entity_types: List of entity types 

60  

61 Returns: 

62 Tuple of (is_proxy, connecting_predicate) 

63 """ 

64 if not entity_types: 

65 return False, "" 

66 

67 display_rules = get_display_rules() 

68 if not display_rules: 

69 return False, "" 

70 

71 for entity_type in entity_types: 

72 for rule in display_rules: 

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

74 intermediate_config = prop.get("intermediateRelation") 

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

76 return True, prop["property"] 

77 

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

79 nested_intermediate_config = display_rule.get("intermediateRelation") 

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

81 return True, prop["property"] 

82 

83 return False, "" 

84 

85 

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

87 """ 

88 Resolve proxy entities to their source entities. 

89  

90 Args: 

91 subject_uri: URI of the proxy entity 

92 predicate: Original predicate 

93 connecting_predicate: The predicate that connects source to proxy 

94  

95 Returns: 

96 Tuple of (final_subject_uri, final_predicate) 

97 """ 

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

99 sparql = get_sparql() 

100 proxy_query_parts = [ 

101 "SELECT DISTINCT ?source WHERE {", 

102 ] 

103 

104 if is_virtuoso: 

105 proxy_query_parts.extend([ 

106 " GRAPH ?g {", 

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

108 " }", 

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

110 ]) 

111 else: 

112 proxy_query_parts.extend([ 

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

114 ]) 

115 

116 proxy_query_parts.append("} LIMIT 1") 

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

118 

119 try: 

120 sparql.setQuery(proxy_query) 

121 sparql.setReturnFormat(JSON) 

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

123 

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

125 if proxy_bindings: 

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

127 

128 return source_uri, connecting_predicate 

129 

130 except Exception as e: 

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

132 

133 return subject_uri, predicate 

134 

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

136 """ 

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

138 

139 Args: 

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

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

142 offset: Number of references to skip. 

143 

144 Returns: 

145 A tuple containing: 

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

147 - Boolean indicating if there are more references. 

148 """ 

149 sparql = get_sparql() 

150 custom_filter = get_custom_filter() 

151 references = [] 

152 query_limit = limit + 1 

153 

154 try: 

155 query_parts = [ 

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

157 ] 

158 if is_virtuoso: 

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

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

161 else: 

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

163 

164 query_parts.extend([ 

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

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

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

168 ]) 

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

170 

171 sparql.setQuery(main_query) 

172 sparql.setReturnFormat(JSON) 

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

174 

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

176 

177 # Determine if there are more results 

178 has_more = len(bindings) > limit 

179 

180 # Process only up to 'limit' results 

181 results_to_process = bindings[:limit] 

182 

183 for result in results_to_process: 

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

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

186 

187 types = get_entity_types(subject) 

188 

189 if _is_virtual_property_intermediate_entity(types): 

190 continue 

191 

192 highest_priority_type = get_highest_priority_class(types) 

193 shape = determine_shape_for_classes(types) 

194 

195 is_proxy, connecting_predicate = _is_proxy_entity(types) 

196 if is_proxy: 

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

198 else: 

199 final_subject, final_predicate = subject, predicate 

200 

201 if final_subject != subject: 

202 final_types = get_entity_types(final_subject) 

203 final_highest_priority_type = get_highest_priority_class(final_types) 

204 final_shape = determine_shape_for_classes(final_types) 

205 else: 

206 final_types = types 

207 final_highest_priority_type = highest_priority_type 

208 final_shape = shape 

209 

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

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

212 

213 references.append({ 

214 "subject": final_subject, 

215 "predicate": final_predicate, 

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

217 "type_label": type_label, 

218 "label": label 

219 }) 

220 

221 return references, has_more 

222 

223 except Exception as e: 

224 tb_str = traceback.format_exc() 

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

226 return [], False 

227 

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

229@login_required 

230def get_linked_resources_api(): 

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

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

233 try: 

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

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

236 except ValueError: 

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

238 

239 if not subject_uri: 

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

241 

242 if limit <= 0 or offset < 0: 

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

244 

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

246 

247 return jsonify({ 

248 "status": "success", 

249 "results": references, 

250 "has_more": has_more 

251 })