Coverage for heritrace / routes / main.py: 99%
75 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-02 10:16 +0000
1# SPDX-FileCopyrightText: 2024-2025 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import json
6import time
8from flask import Blueprint, current_app, redirect, render_template, request, url_for
9from flask_login import login_required
10from SPARQLWrapper import JSON
11from werkzeug.wrappers import Response as WerkzeugResponse
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 get_sortable_properties,
22)
24main_bp = Blueprint("main", __name__)
27@main_bp.route("/")
28def index() -> str:
29 return render_template("index.jinja")
32@main_bp.route("/catalogue")
33@login_required
34def catalogue() -> str:
35 page = int(request.args.get("page", 1))
36 per_page = int(
37 request.args.get("per_page", current_app.config["CATALOGUE_DEFAULT_PER_PAGE"])
38 )
39 selected_class = request.args.get("class")
40 sort_property = request.args.get("sort_property")
41 sort_direction = request.args.get("sort_direction", "ASC")
42 selected_shape = request.args.get("shape")
44 available_classes = get_available_classes()
46 if not selected_class and available_classes:
47 selected_class = str(available_classes[0]["uri"])
49 if not selected_shape and selected_class:
50 selected_shape = determine_shape_for_classes([selected_class])
52 catalog_data = get_catalog_data(
53 CatalogQuery(
54 selected_class=selected_class,
55 page=page,
56 per_page=per_page,
57 sort_property=sort_property,
58 sort_direction=sort_direction,
59 selected_shape=selected_shape,
60 ),
61 available_classes,
62 )
64 return render_template(
65 "catalogue.jinja",
66 available_classes=available_classes,
67 selected_class=selected_class,
68 selected_shape=selected_shape,
69 page=page,
70 total_entity_pages=catalog_data["total_pages"],
71 per_page=per_page,
72 allowed_per_page=current_app.config["CATALOGUE_ALLOWED_PER_PAGE"],
73 sortable_properties=json.dumps(catalog_data["sortable_properties"]),
74 current_sort_property=catalog_data["sort_property"],
75 current_sort_direction=catalog_data["sort_direction"],
76 initial_entities=catalog_data["entities"],
77 )
80@main_bp.route("/time-vault")
81@login_required
82def time_vault() -> str:
83 """
84 Render the Time Vault page, which displays a list of deleted entities.
85 """
86 initial_page = request.args.get("page", 1, type=int)
87 initial_per_page = request.args.get(
88 "per_page", current_app.config["CATALOGUE_DEFAULT_PER_PAGE"], type=int
89 )
90 sort_property = request.args.get("sort_property", "deletionTime")
91 sort_direction = request.args.get("sort_direction", "DESC")
92 selected_class = request.args.get("class")
93 selected_shape = request.args.get("shape")
95 allowed_per_page = current_app.config["CATALOGUE_ALLOWED_PER_PAGE"]
97 (
98 initial_entities,
99 available_classes,
100 selected_class,
101 selected_shape,
102 sortable_properties,
103 total_count,
104 ) = get_deleted_entities_with_filtering(
105 DeletedEntitiesQuery(
106 initial_page,
107 initial_per_page,
108 sort_property,
109 sort_direction,
110 selected_class,
111 selected_shape,
112 )
113 )
115 sortable_properties = [
116 {"property": "deletionTime", "displayName": "Deletion Time", "sortType": "date"}
117 ]
119 if selected_class is not None:
120 entity_key = (selected_class, selected_shape)
121 sortable_properties.extend(get_sortable_properties(entity_key))
123 sortable_properties = json.dumps(sortable_properties)
125 return render_template(
126 "time_vault.jinja",
127 available_classes=available_classes,
128 selected_class=selected_class,
129 selected_shape=selected_shape,
130 page=initial_page,
131 total_entity_pages=(total_count + initial_per_page - 1) // initial_per_page
132 if total_count > 0
133 else 0,
134 per_page=initial_per_page,
135 allowed_per_page=allowed_per_page,
136 sortable_properties=sortable_properties,
137 current_sort_property=sort_property,
138 current_sort_direction=sort_direction,
139 initial_entities=initial_entities,
140 )
143@main_bp.route("/dataset-endpoint", methods=["POST"])
144@login_required
145def sparql_proxy() -> tuple[str, int, dict[str, str]]:
146 query = request.form.get("query", "")
148 sparql_wrapper = get_sparql()
149 sparql_wrapper.setQuery(query)
150 sparql_wrapper.setReturnFormat(JSON)
152 # Implement retry mechanism
153 max_retries = 3
154 retry_delay = 1 # seconds
156 for attempt in range(max_retries):
157 try:
158 results = sparql_wrapper.query().convert()
159 return (
160 json.dumps(results),
161 200,
162 {"Content-Type": "application/sparql-results+json"},
163 )
164 except Exception as e: # noqa: PERF203
165 if attempt < max_retries - 1:
166 time.sleep(retry_delay)
167 retry_delay *= 2 # Exponential backoff
168 else:
169 current_app.logger.exception(
170 "All SPARQL query attempts failed for query: %s", query
171 )
172 return (
173 json.dumps({"error": str(e)}),
174 500,
175 {"Content-Type": "application/json"},
176 )
178 return (
179 json.dumps({"error": "All SPARQL query attempts failed"}),
180 500,
181 {"Content-Type": "application/json"},
182 )
185@main_bp.route("/endpoint")
186@login_required
187def endpoint() -> str:
188 return render_template("endpoint.jinja", dataset_endpoint=get_dataset_endpoint())
191@main_bp.route("/search")
192@login_required
193def search() -> WerkzeugResponse:
194 subject = request.args.get("q")
195 return redirect(url_for("entity.about", subject=subject))