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

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_proxy_entity(entity_types: list) -> tuple[bool, str]: 

19 """ 

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

21  

22 Args: 

23 entity_types: List of entity types 

24  

25 Returns: 

26 Tuple of (is_proxy, connecting_predicate) 

27 """ 

28 if not entity_types: 

29 return False, "" 

30 

31 display_rules = get_display_rules() 

32 if not display_rules: 

33 return False, "" 

34 

35 for entity_type in entity_types: 

36 for rule in display_rules: 

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

38 intermediate_config = prop.get("intermediateRelation") 

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

40 return True, prop["property"] 

41 

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

43 nested_intermediate_config = display_rule.get("intermediateRelation") 

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

45 return True, prop["property"] 

46 

47 return False, "" 

48 

49 

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

51 """ 

52 Resolve proxy entities to their source entities. 

53  

54 Args: 

55 subject_uri: URI of the proxy entity 

56 predicate: Original predicate 

57 connecting_predicate: The predicate that connects source to proxy 

58  

59 Returns: 

60 Tuple of (final_subject_uri, final_predicate) 

61 """ 

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

63 sparql = get_sparql() 

64 proxy_query_parts = [ 

65 "SELECT DISTINCT ?source WHERE {", 

66 ] 

67 

68 if is_virtuoso: 

69 proxy_query_parts.extend([ 

70 " GRAPH ?g {", 

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

72 " }", 

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

74 ]) 

75 else: 

76 proxy_query_parts.extend([ 

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

78 ]) 

79 

80 proxy_query_parts.append("} LIMIT 1") 

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

82 

83 try: 

84 sparql.setQuery(proxy_query) 

85 sparql.setReturnFormat(JSON) 

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

87 

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

89 if proxy_bindings: 

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

91 

92 return source_uri, connecting_predicate 

93 

94 except Exception as e: 

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

96 

97 return subject_uri, predicate 

98 

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

100 """ 

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

102 

103 Args: 

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

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

106 offset: Number of references to skip. 

107 

108 Returns: 

109 A tuple containing: 

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

111 - Boolean indicating if there are more references. 

112 """ 

113 sparql = get_sparql() 

114 custom_filter = get_custom_filter() 

115 references = [] 

116 query_limit = limit + 1 

117 

118 try: 

119 query_parts = [ 

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

121 ] 

122 if is_virtuoso: 

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

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

125 else: 

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

127 

128 query_parts.extend([ 

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

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

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

132 ]) 

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

134 

135 sparql.setQuery(main_query) 

136 sparql.setReturnFormat(JSON) 

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

138 

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

140 

141 # Determine if there are more results 

142 has_more = len(bindings) > limit 

143 

144 # Process only up to 'limit' results 

145 results_to_process = bindings[:limit] 

146 

147 for result in results_to_process: 

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

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

150 

151 types = get_entity_types(subject) 

152 highest_priority_type = get_highest_priority_class(types) 

153 shape = determine_shape_for_classes(types) 

154 

155 is_proxy, connecting_predicate = _is_proxy_entity(types) 

156 if is_proxy: 

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

158 else: 

159 final_subject, final_predicate = subject, predicate 

160 

161 if final_subject != subject: 

162 final_types = get_entity_types(final_subject) 

163 final_highest_priority_type = get_highest_priority_class(final_types) 

164 final_shape = determine_shape_for_classes(final_types) 

165 else: 

166 final_types = types 

167 final_highest_priority_type = highest_priority_type 

168 final_shape = shape 

169 

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

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

172 

173 references.append({ 

174 "subject": final_subject, 

175 "predicate": final_predicate, 

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

177 "type_label": type_label, 

178 "label": label 

179 }) 

180 

181 return references, has_more 

182 

183 except Exception as e: 

184 tb_str = traceback.format_exc() 

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

186 return [], False 

187 

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

189@login_required 

190def get_linked_resources_api(): 

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

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

193 try: 

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

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

196 except ValueError: 

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

198 

199 if not subject_uri: 

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

201 

202 if limit <= 0 or offset < 0: 

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

204 

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

206 

207 return jsonify({ 

208 "status": "success", 

209 "results": references, 

210 "has_more": has_more 

211 })