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

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5import json 

6 

7from flask import Blueprint, current_app, redirect, render_template, request, url_for 

8from flask_login import login_required 

9from SPARQLWrapper import JSON 

10from SPARQLWrapper.SPARQLExceptions import SPARQLWrapperException 

11from werkzeug.wrappers import Response as WerkzeugResponse 

12 

13from heritrace.extensions import get_dataset_endpoint, get_sparql 

14from heritrace.utils.shacl_utils import determine_shape_for_classes 

15from heritrace.utils.sparql_utils import ( 

16 CatalogQuery, 

17 DeletedEntitiesQuery, 

18 get_available_classes, 

19 get_catalog_data, 

20 get_deleted_entities_with_filtering, 

21) 

22 

23main_bp = Blueprint("main", __name__) 

24 

25 

26@main_bp.route("/") 

27def index() -> str: 

28 return render_template("index.jinja") 

29 

30 

31@main_bp.route("/catalogue") 

32@login_required 

33def catalogue() -> str: 

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

35 per_page = int( 

36 request.args.get("per_page", current_app.config["CATALOGUE_DEFAULT_PER_PAGE"]) 

37 ) 

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

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

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

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

42 

43 available_classes = get_available_classes() 

44 

45 if not selected_class and available_classes: 

46 selected_class = str(available_classes[0]["uri"]) 

47 

48 if not selected_shape and selected_class: 

49 selected_shape = determine_shape_for_classes([selected_class]) 

50 

51 catalog_data = get_catalog_data( 

52 CatalogQuery( 

53 selected_class=selected_class, 

54 page=page, 

55 per_page=per_page, 

56 sort_property=sort_property, 

57 sort_direction=sort_direction, 

58 selected_shape=selected_shape, 

59 ), 

60 available_classes, 

61 ) 

62 

63 return render_template( 

64 "catalogue.jinja", 

65 available_classes=available_classes, 

66 selected_class=selected_class, 

67 selected_shape=selected_shape, 

68 page=page, 

69 total_entity_pages=catalog_data["total_pages"], 

70 per_page=per_page, 

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

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

73 current_sort_property=catalog_data["sort_property"], 

74 current_sort_direction=catalog_data["sort_direction"], 

75 initial_entities=catalog_data["entities"], 

76 ) 

77 

78 

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

80@login_required 

81def time_vault() -> str: 

82 """ 

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

84 """ 

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

86 initial_per_page = request.args.get( 

87 "per_page", current_app.config["CATALOGUE_DEFAULT_PER_PAGE"], type=int 

88 ) 

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

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

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

92 

93 allowed_per_page = current_app.config["CATALOGUE_ALLOWED_PER_PAGE"] 

94 

95 ( 

96 initial_entities, 

97 available_classes, 

98 selected_class, 

99 selected_shape, 

100 sortable_properties, 

101 total_count, 

102 ) = get_deleted_entities_with_filtering( 

103 DeletedEntitiesQuery( 

104 initial_page, 

105 initial_per_page, 

106 sort_direction, 

107 selected_class, 

108 selected_shape, 

109 ) 

110 ) 

111 

112 return render_template( 

113 "time_vault.jinja", 

114 available_classes=available_classes, 

115 selected_class=selected_class, 

116 selected_shape=selected_shape, 

117 page=initial_page, 

118 total_entity_pages=(total_count + initial_per_page - 1) // initial_per_page 

119 if total_count > 0 

120 else 0, 

121 per_page=initial_per_page, 

122 allowed_per_page=allowed_per_page, 

123 sortable_properties=json.dumps(sortable_properties), 

124 current_sort_property=sortable_properties[0]["property"], 

125 current_sort_direction=sort_direction, 

126 initial_entities=initial_entities, 

127 ) 

128 

129 

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

131@login_required 

132def sparql_proxy() -> tuple[str, int, dict[str, str]]: 

133 query = request.form.get("query", "") 

134 

135 sparql_wrapper = get_sparql() 

136 sparql_wrapper.setQuery(query) 

137 sparql_wrapper.setReturnFormat(JSON) 

138 

139 try: 

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

141 return ( 

142 json.dumps(results), 

143 200, 

144 {"Content-Type": "application/sparql-results+json"}, 

145 ) 

146 except ( 

147 json.JSONDecodeError, 

148 OSError, 

149 SPARQLWrapperException, 

150 TypeError, 

151 UnicodeDecodeError, 

152 ) as error: 

153 current_app.logger.exception("SPARQL query failed for query: %s", query) 

154 return ( 

155 json.dumps({"error": str(error)}), 

156 500, 

157 {"Content-Type": "application/json"}, 

158 ) 

159 

160 

161@main_bp.route("/endpoint") 

162@login_required 

163def endpoint() -> str: 

164 return render_template("endpoint.jinja", dataset_endpoint=get_dataset_endpoint()) 

165 

166 

167@main_bp.route("/search") 

168@login_required 

169def search() -> WerkzeugResponse: 

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

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