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

1import json 

2import time 

3 

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

5 url_for) 

6from flask_login import login_required 

7from heritrace.extensions import get_sparql 

8from heritrace.utils.shacl_utils import determine_shape_for_classes 

9from heritrace.utils.sparql_utils import (get_available_classes, 

10 get_catalog_data, 

11 get_deleted_entities_with_filtering, 

12 get_sortable_properties) 

13from SPARQLWrapper import JSON 

14 

15main_bp = Blueprint("main", __name__) 

16 

17 

18@main_bp.route("/") 

19def index(): 

20 return render_template("index.jinja") 

21 

22 

23@main_bp.route("/catalogue") 

24@login_required 

25def catalogue(): 

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

27 per_page = int(request.args.get("per_page", 50)) 

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

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

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

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

32 

33 available_classes = get_available_classes() 

34 

35 if not selected_class and available_classes: 

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

37 

38 if not selected_shape and selected_class: 

39 selected_shape = determine_shape_for_classes([selected_class]) 

40 

41 catalog_data = get_catalog_data( 

42 selected_class, 

43 page, 

44 per_page, 

45 sort_property, 

46 sort_direction, 

47 selected_shape 

48 ) 

49 

50 return render_template( 

51 "catalogue.jinja", 

52 available_classes=available_classes, 

53 selected_class=selected_class, 

54 selected_shape=selected_shape, 

55 page=page, 

56 total_entity_pages=catalog_data["total_pages"], 

57 per_page=per_page, 

58 allowed_per_page=[50, 100, 200, 500], 

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

60 current_sort_property=catalog_data["sort_property"], 

61 current_sort_direction=catalog_data["sort_direction"], 

62 initial_entities=catalog_data["entities"], 

63 ) 

64 

65 

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

67@login_required 

68def time_vault(): 

69 """ 

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

71 """ 

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

73 initial_per_page = request.args.get("per_page", 50, type=int) 

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

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

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

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

78 

79 allowed_per_page = [50, 100, 200, 500] 

80 

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

82 get_deleted_entities_with_filtering( 

83 initial_page, 

84 initial_per_page, 

85 sort_property, 

86 sort_direction, 

87 selected_class, 

88 selected_shape 

89 ) 

90 ) 

91 

92 sortable_properties = [ 

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

94 ] 

95 

96 if selected_class is not None: 

97 entity_key = (selected_class, selected_shape) 

98 sortable_properties.extend( 

99 get_sortable_properties(entity_key) 

100 ) 

101 

102 sortable_properties = json.dumps(sortable_properties) 

103 

104 return render_template( 

105 "time_vault.jinja", 

106 available_classes=available_classes, 

107 selected_class=selected_class, 

108 selected_shape=selected_shape, 

109 page=initial_page, 

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

111 per_page=initial_per_page, 

112 allowed_per_page=allowed_per_page, 

113 sortable_properties=sortable_properties, 

114 current_sort_property=sort_property, 

115 current_sort_direction=sort_direction, 

116 initial_entities=initial_entities, 

117 ) 

118 

119 

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

121@login_required 

122def sparql_proxy(): 

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

124 

125 # Use SPARQLWrapper instead of direct requests 

126 sparql_wrapper = get_sparql() 

127 sparql_wrapper.setQuery(query) 

128 sparql_wrapper.setReturnFormat(JSON) 

129 

130 # Implement retry mechanism 

131 max_retries = 3 

132 retry_delay = 1 # seconds 

133 

134 for attempt in range(max_retries): 

135 try: 

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

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

138 except Exception as e: 

139 if attempt < max_retries - 1: 

140 time.sleep(retry_delay) 

141 retry_delay *= 2 # Exponential backoff 

142 else: 

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

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

145 

146 

147@main_bp.route("/endpoint") 

148@login_required 

149def endpoint(): 

150 from heritrace.extensions import dataset_endpoint 

151 

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

153 

154 

155@main_bp.route("/search") 

156@login_required 

157def search(): 

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

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