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

1# SPDX-FileCopyrightText: 2024-2025 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5import json 

6import time 

7 

8from flask import (Blueprint, current_app, redirect, render_template, request, 

9 url_for) 

10from flask_login import login_required 

11from heritrace.extensions import get_sparql 

12from heritrace.utils.shacl_utils import determine_shape_for_classes 

13from heritrace.utils.sparql_utils import (get_available_classes, 

14 get_catalog_data, 

15 get_deleted_entities_with_filtering, 

16 get_sortable_properties) 

17from SPARQLWrapper import JSON 

18 

19main_bp = Blueprint("main", __name__) 

20 

21 

22@main_bp.route("/") 

23def index(): 

24 return render_template("index.jinja") 

25 

26 

27@main_bp.route("/catalogue") 

28@login_required 

29def catalogue(): 

30 page = int(request.args.get("page", 1)) 

31 per_page = int(request.args.get("per_page", current_app.config["CATALOGUE_DEFAULT_PER_PAGE"])) 

32 selected_class = request.args.get("class") 

33 sort_property = request.args.get("sort_property") 

34 sort_direction = request.args.get("sort_direction", "ASC") 

35 selected_shape = request.args.get("shape") 

36 

37 available_classes = get_available_classes() 

38 

39 if not selected_class and available_classes: 

40 selected_class = available_classes[0]["uri"] 

41 

42 if not selected_shape and selected_class: 

43 selected_shape = determine_shape_for_classes([selected_class]) 

44 

45 catalog_data = get_catalog_data( 

46 selected_class, 

47 page, 

48 per_page, 

49 sort_property, 

50 sort_direction, 

51 selected_shape 

52 ) 

53 

54 return render_template( 

55 "catalogue.jinja", 

56 available_classes=available_classes, 

57 selected_class=selected_class, 

58 selected_shape=selected_shape, 

59 page=page, 

60 total_entity_pages=catalog_data["total_pages"], 

61 per_page=per_page, 

62 allowed_per_page=current_app.config["CATALOGUE_ALLOWED_PER_PAGE"], 

63 sortable_properties=json.dumps(catalog_data["sortable_properties"]), 

64 current_sort_property=catalog_data["sort_property"], 

65 current_sort_direction=catalog_data["sort_direction"], 

66 initial_entities=catalog_data["entities"], 

67 ) 

68 

69 

70@main_bp.route("/time-vault") 

71@login_required 

72def time_vault(): 

73 """ 

74 Render the Time Vault page, which displays a list of deleted entities. 

75 """ 

76 initial_page = request.args.get("page", 1, type=int) 

77 initial_per_page = request.args.get("per_page", current_app.config["CATALOGUE_DEFAULT_PER_PAGE"], type=int) 

78 sort_property = request.args.get("sort_property", "deletionTime") 

79 sort_direction = request.args.get("sort_direction", "DESC") 

80 selected_class = request.args.get("class") 

81 selected_shape = request.args.get("shape") 

82 

83 allowed_per_page = current_app.config["CATALOGUE_ALLOWED_PER_PAGE"] 

84 

85 initial_entities, available_classes, selected_class, selected_shape, sortable_properties, total_count = ( 

86 get_deleted_entities_with_filtering( 

87 initial_page, 

88 initial_per_page, 

89 sort_property, 

90 sort_direction, 

91 selected_class, 

92 selected_shape 

93 ) 

94 ) 

95 

96 sortable_properties = [ 

97 {"property": "deletionTime", "displayName": "Deletion Time", "sortType": "date"} 

98 ] 

99 

100 if selected_class is not None: 

101 entity_key = (selected_class, selected_shape) 

102 sortable_properties.extend( 

103 get_sortable_properties(entity_key) 

104 ) 

105 

106 sortable_properties = json.dumps(sortable_properties) 

107 

108 return render_template( 

109 "time_vault.jinja", 

110 available_classes=available_classes, 

111 selected_class=selected_class, 

112 selected_shape=selected_shape, 

113 page=initial_page, 

114 total_entity_pages=(total_count + initial_per_page - 1) // initial_per_page if total_count > 0 else 0, 

115 per_page=initial_per_page, 

116 allowed_per_page=allowed_per_page, 

117 sortable_properties=sortable_properties, 

118 current_sort_property=sort_property, 

119 current_sort_direction=sort_direction, 

120 initial_entities=initial_entities, 

121 ) 

122 

123 

124@main_bp.route("/dataset-endpoint", methods=["POST"]) 

125@login_required 

126def sparql_proxy(): 

127 query = request.form.get("query") 

128 

129 # Use SPARQLWrapper instead of direct requests 

130 sparql_wrapper = get_sparql() 

131 sparql_wrapper.setQuery(query) 

132 sparql_wrapper.setReturnFormat(JSON) 

133 

134 # Implement retry mechanism 

135 max_retries = 3 

136 retry_delay = 1 # seconds 

137 

138 for attempt in range(max_retries): 

139 try: 

140 results = sparql_wrapper.query().convert() 

141 return json.dumps(results), 200, {"Content-Type": "application/sparql-results+json"} 

142 except Exception as e: 

143 if attempt < max_retries - 1: 

144 time.sleep(retry_delay) 

145 retry_delay *= 2 # Exponential backoff 

146 else: 

147 current_app.logger.error(f"All SPARQL query attempts failed for query: {query}") 

148 return json.dumps({"error": str(e)}), 500, {"Content-Type": "application/json"} 

149 

150 

151@main_bp.route("/endpoint") 

152@login_required 

153def endpoint(): 

154 from heritrace.extensions import dataset_endpoint 

155 

156 return render_template("endpoint.jinja", dataset_endpoint=dataset_endpoint) 

157 

158 

159@main_bp.route("/search") 

160@login_required 

161def search(): 

162 subject = request.args.get("q") 

163 return redirect(url_for("entity.about", subject=subject))