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_sparql 

7from heritrace.utils.display_rules_utils import get_highest_priority_class 

8from heritrace.utils.shacl_utils import determine_shape_for_classes 

9from heritrace.utils.sparql_utils import get_entity_types 

10from heritrace.utils.virtuoso_utils import (VIRTUOSO_EXCLUDED_GRAPHS, 

11 is_virtuoso) 

12from SPARQLWrapper import JSON 

13 

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

15 

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

17 """ 

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

19 

20 Args: 

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

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

23 offset: Number of references to skip. 

24 

25 Returns: 

26 A tuple containing: 

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

28 - Boolean indicating if there are more references. 

29 """ 

30 sparql = get_sparql() 

31 custom_filter = get_custom_filter() 

32 references = [] 

33 # Fetch limit + 1 to check if there are more results 

34 query_limit = limit + 1 

35 

36 try: 

37 # Main Query with pagination (limit + 1) 

38 query_parts = [ 

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

40 ] 

41 if is_virtuoso: 

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

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

44 else: 

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

46 

47 query_parts.extend([ 

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

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

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

51 ]) 

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

53 

54 sparql.setQuery(main_query) 

55 sparql.setReturnFormat(JSON) 

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

57 

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

59 

60 # Determine if there are more results 

61 has_more = len(bindings) > limit 

62 

63 # Process only up to 'limit' results 

64 results_to_process = bindings[:limit] 

65 

66 for result in results_to_process: 

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

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

69 

70 types = get_entity_types(subject) 

71 highest_priority_type = get_highest_priority_class(types) 

72 shape = determine_shape_for_classes(types) 

73 label = custom_filter.human_readable_entity(subject, (highest_priority_type, shape)) 

74 type_label = custom_filter.human_readable_class((highest_priority_type, shape)) if highest_priority_type else None 

75 

76 references.append({ 

77 "subject": subject, 

78 "predicate": predicate, 

79 "predicate_label": custom_filter.human_readable_predicate(predicate, (highest_priority_type, shape)), 

80 "type_label": type_label, 

81 "label": label 

82 }) 

83 

84 return references, has_more 

85 

86 except Exception as e: 

87 tb_str = traceback.format_exc() 

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

89 return [], False 

90 

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

92@login_required 

93def get_linked_resources_api(): 

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

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

96 try: 

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

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

99 except ValueError: 

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

101 

102 if not subject_uri: 

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

104 

105 if limit <= 0 or offset < 0: 

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

107 

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

109 

110 return jsonify({ 

111 "status": "success", 

112 "results": references, 

113 "has_more": has_more 

114 })