Coverage for heritrace/routes/api.py: 99%
551 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 08:34 +0000
1# SPDX-FileCopyrightText: 2024-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import traceback
6from dataclasses import dataclass
7from typing import TypedDict, cast
9from flask import (
10 Blueprint,
11 Response,
12 current_app,
13 g,
14 jsonify,
15 render_template_string,
16 request,
17)
18from flask_babel import gettext
19from flask_login import current_user, login_required
20from rdflib import RDF, XSD, Graph, Literal, URIRef
22from heritrace.apis.orcid import get_responsible_agent_uri
23from heritrace.editor import Editor, EndpointConfig
24from heritrace.extensions import (
25 get_custom_filter,
26 get_dataset_endpoint,
27 get_form_fields,
28 get_provenance_endpoint,
29)
30from heritrace.services.resource_lock_manager import LockStatus
31from heritrace.utils.datatypes import DATATYPE_MAPPING
32from heritrace.utils.primary_source_utils import save_user_default_primary_source
33from heritrace.utils.shacl_utils import determine_shape_for_classes
34from heritrace.utils.shacl_validation import validate_new_triple
35from heritrace.utils.sparql_utils import (
36 CatalogQuery,
37 DeletedEntitiesQuery,
38 find_orphaned_entities,
39 get_available_classes,
40 get_catalog_data,
41 get_deleted_entities_with_filtering,
42 get_triples_from_graph,
43 import_entity_graph,
44 import_referenced_entities,
45)
46from heritrace.utils.strategies import OrphanHandlingStrategy, ProxyHandlingStrategy
47from heritrace.utils.uri_utils import generate_unique_uri, is_valid_url
48from heritrace.utils.virtual_properties import transform_changes_with_virtual_properties
51@dataclass(frozen=True, slots=True)
52class ChangeOperation:
53 editor: Editor
54 subject: URIRef
55 graph_uri: URIRef | None = None
56 entity_type: str | None = None
57 entity_shape: str | None = None
60api_bp = Blueprint("api", __name__)
63@api_bp.route("/catalogue")
64@login_required
65def catalogue_api() -> Response:
66 selected_class = request.args.get("class")
67 selected_shape = request.args.get("shape")
68 page = int(request.args.get("page", 1))
69 per_page = int(
70 request.args.get("per_page", current_app.config["CATALOGUE_DEFAULT_PER_PAGE"])
71 )
72 sort_property = request.args.get("sort_property")
73 sort_direction = request.args.get("sort_direction", "ASC")
75 allowed_per_page = current_app.config["CATALOGUE_ALLOWED_PER_PAGE"]
76 if per_page not in allowed_per_page:
77 per_page = current_app.config["CATALOGUE_DEFAULT_PER_PAGE"]
79 if not sort_property or sort_property.lower() == "null":
80 sort_property = None
82 available_classes = get_available_classes()
84 catalog_data = get_catalog_data(
85 CatalogQuery(
86 selected_class=selected_class,
87 page=page,
88 per_page=per_page,
89 sort_property=sort_property,
90 sort_direction=sort_direction,
91 selected_shape=selected_shape,
92 ),
93 available_classes,
94 )
96 catalog_data["available_classes"] = available_classes
97 return jsonify(catalog_data)
100@api_bp.route("/time-vault")
101@login_required
102def get_deleted_entities_api() -> Response:
103 """
104 API endpoint to retrieve deleted entities with pagination and sorting.
105 Only processes and returns entities whose classes are marked as visible.
106 """
107 selected_class = request.args.get("class")
108 selected_shape = request.args.get("shape")
109 page = int(request.args.get("page", 1))
110 per_page = int(
111 request.args.get("per_page", current_app.config["CATALOGUE_DEFAULT_PER_PAGE"])
112 )
113 sort_direction = request.args.get("sort_direction", "DESC")
115 allowed_per_page = current_app.config["CATALOGUE_ALLOWED_PER_PAGE"]
116 if per_page not in allowed_per_page:
117 per_page = current_app.config["CATALOGUE_DEFAULT_PER_PAGE"]
119 (
120 deleted_entities,
121 available_classes,
122 selected_class,
123 selected_shape,
124 sortable_properties,
125 total_count,
126 ) = get_deleted_entities_with_filtering(
127 DeletedEntitiesQuery(
128 page,
129 per_page,
130 sort_direction,
131 selected_class,
132 selected_shape,
133 )
134 )
136 return jsonify(
137 {
138 "entities": deleted_entities,
139 "total_pages": (total_count + per_page - 1) // per_page
140 if total_count > 0
141 else 0,
142 "current_page": page,
143 "per_page": per_page,
144 "total_count": total_count,
145 "sort_property": sortable_properties[0]["property"],
146 "sort_direction": sort_direction,
147 "selected_class": selected_class,
148 "selected_shape": selected_shape,
149 "available_classes": available_classes,
150 "sortable_properties": sortable_properties,
151 }
152 )
155@api_bp.route("/check-lock", methods=["POST"])
156@login_required
157def check_lock() -> Response | tuple[Response, int]:
158 """Check if a resource is locked."""
159 try:
160 data = request.get_json()
161 resource_uri = data.get("resource_uri")
163 if not resource_uri:
164 return (
165 jsonify(
166 {"status": "error", "message": gettext("No resource URI provided")}
167 ),
168 400,
169 )
171 status, lock_info = g.resource_lock_manager.check_lock_status(resource_uri)
173 if status == LockStatus.LOCKED:
174 return jsonify(
175 {
176 "status": "locked",
177 "title": gettext("Resource Locked"),
178 "message": gettext(
179 "This resource is currently being"
180 " edited by %(user)s [%(orcid)s]",
181 user=lock_info.user_name,
182 orcid=lock_info.user_id,
183 ),
184 }
185 )
186 if status == LockStatus.ERROR:
187 return (
188 jsonify(
189 {
190 "status": "error",
191 "title": gettext("Error"),
192 "message": gettext("An error occurred while checking the lock"),
193 }
194 ),
195 500,
196 )
197 return jsonify({"status": "available"})
199 except Exception:
200 current_app.logger.exception("Error in check_lock")
201 return (
202 jsonify(
203 {
204 "status": "error",
205 "title": gettext("Error"),
206 "message": gettext("An unexpected error occurred"),
207 }
208 ),
209 500,
210 )
213@api_bp.route("/acquire-lock", methods=["POST"])
214@login_required
215def acquire_lock() -> Response | tuple[Response, int]:
216 """Try to acquire a lock on a resource."""
217 try:
218 data = request.get_json()
219 resource_uri = data.get("resource_uri")
220 linked_resources = data.get("linked_resources", [])
222 if not resource_uri:
223 return (
224 jsonify(
225 {"status": "error", "message": gettext("No resource URI provided")}
226 ),
227 400,
228 )
230 # First check if the resource or any related resource is locked by another user
231 status, lock_info = g.resource_lock_manager.check_lock_status(resource_uri)
232 if status == LockStatus.LOCKED:
233 return (
234 jsonify(
235 {
236 "status": "locked",
237 "title": gettext("Resource Locked"),
238 "message": gettext(
239 "This resource is currently"
240 " being edited by"
241 " %(user)s [%(orcid)s]",
242 user=lock_info.user_name,
243 orcid=lock_info.user_id,
244 ),
245 }
246 ),
247 200,
248 )
250 # Use the provided linked_resources
251 success = g.resource_lock_manager.acquire_lock(resource_uri, linked_resources)
253 if success:
254 return jsonify({"status": "success"})
256 return (
257 jsonify(
258 {
259 "status": "error",
260 "message": gettext("Resource is locked by another user"),
261 }
262 ),
263 423,
264 )
266 except Exception:
267 current_app.logger.exception("Error in acquire_lock")
268 return (
269 jsonify(
270 {"status": "error", "message": gettext("An unexpected error occurred")}
271 ),
272 500,
273 )
276@api_bp.route("/release-lock", methods=["POST"])
277@login_required
278def release_lock() -> Response | tuple[Response, int]:
279 """Release a lock on a resource."""
280 try:
281 data = request.get_json()
282 resource_uri = data.get("resource_uri")
284 if not resource_uri:
285 return (
286 jsonify(
287 {"status": "error", "message": gettext("No resource URI provided")}
288 ),
289 400,
290 )
292 success = g.resource_lock_manager.release_lock(resource_uri)
294 if success:
295 return jsonify({"status": "success"})
297 return (
298 jsonify({"status": "error", "message": gettext("Unable to release lock")}),
299 400,
300 )
302 except Exception:
303 current_app.logger.exception("Error in release_lock")
304 return (
305 jsonify(
306 {"status": "error", "message": gettext("An unexpected error occurred")}
307 ),
308 500,
309 )
312@api_bp.route("/renew-lock", methods=["POST"])
313@login_required
314def renew_lock() -> Response | tuple[Response, int]:
315 """Renew an existing lock on a resource."""
316 try:
317 data = request.get_json()
318 resource_uri = data.get("resource_uri")
320 if not resource_uri:
321 return (
322 jsonify(
323 {"status": "error", "message": gettext("No resource URI provided")}
324 ),
325 400,
326 )
328 # When renewing a lock, we don't need to check for linked resources again
329 # Just pass an empty list as we're only refreshing the existing lock
330 success = g.resource_lock_manager.acquire_lock(resource_uri, [])
332 if success:
333 return jsonify({"status": "success"})
335 return (
336 jsonify({"status": "error", "message": gettext("Unable to renew lock")}),
337 423,
338 )
340 except Exception:
341 current_app.logger.exception("Error in renew_lock")
342 return (
343 jsonify(
344 {"status": "error", "message": gettext("An unexpected error occurred")}
345 ),
346 500,
347 )
350@api_bp.route("/validate-literal", methods=["POST"])
351@login_required
352def validate_literal() -> tuple[Response, int]:
353 """Validate a literal value and suggest appropriate datatypes."""
354 value = request.json.get("value")
355 if not value:
356 return jsonify({"error": gettext("Value is required.")}), 400
358 matching_datatypes = []
359 for datatype, validation_func, _ in DATATYPE_MAPPING:
360 if validation_func(value):
361 matching_datatypes.append(str(datatype))
363 if not matching_datatypes:
364 return jsonify({"error": gettext("No matching datatypes found.")}), 400
366 return jsonify({"valid_datatypes": matching_datatypes}), 200
369def _collect_affected_entities(
370 changes: list[dict],
371 entity_type: str,
372 *,
373 check_for_orphans: bool,
374 check_for_proxies: bool,
375) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
376 orphans: list[dict[str, str]] = []
377 intermediate_orphans: list[dict[str, str]] = []
378 for change in changes:
379 if change["action"] == "delete":
380 found_orphans, found_intermediates = find_orphaned_entities(
381 URIRef(change["subject"]),
382 entity_type,
383 URIRef(change["predicate"]) if change.get("predicate") else None,
384 change.get("object"),
385 )
386 if check_for_orphans:
387 orphans.extend(found_orphans)
388 if check_for_proxies:
389 intermediate_orphans.extend(found_intermediates)
390 return orphans, intermediate_orphans
393def _format_orphan_response(
394 orphans: list[dict[str, str]],
395 intermediate_orphans: list[dict[str, str]],
396 entity_shape: str | None,
397 orphan_strategy: OrphanHandlingStrategy,
398 proxy_strategy: ProxyHandlingStrategy,
399) -> Response:
400 custom_filter = get_custom_filter()
402 def format_entities(
403 entities: list[dict[str, str]],
404 *,
405 is_intermediate: bool = False,
406 ) -> list[dict[str, str | bool]]:
407 return [
408 {
409 "uri": entity["uri"],
410 "label": custom_filter.human_readable_entity(
411 entity["uri"], (entity["type"], entity_shape)
412 ),
413 "type": custom_filter.human_readable_class(
414 (entity["type"], entity_shape)
415 ),
416 "is_intermediate": is_intermediate,
417 }
418 for entity in entities
419 ]
421 affected_entities = format_entities(orphans) + format_entities(
422 intermediate_orphans, is_intermediate=True
423 )
425 should_delete = (
426 orphan_strategy == OrphanHandlingStrategy.DELETE
427 and proxy_strategy == ProxyHandlingStrategy.DELETE
428 )
430 return jsonify(
431 {
432 "status": "success",
433 "affected_entities": affected_entities,
434 "should_delete": should_delete,
435 "orphan_strategy": orphan_strategy.value,
436 "proxy_strategy": proxy_strategy.value,
437 }
438 )
441@api_bp.route("/check_orphans", methods=["POST"])
442@login_required
443def check_orphans() -> Response | tuple[Response, int]:
444 try:
445 orphan_strategy = current_app.config.get(
446 "ORPHAN_HANDLING_STRATEGY", OrphanHandlingStrategy.KEEP
447 )
448 proxy_strategy = current_app.config.get(
449 "PROXY_HANDLING_STRATEGY", ProxyHandlingStrategy.KEEP
450 )
452 data = request.json
453 if not data or "changes" not in data or "entity_type" not in data:
454 return (
455 jsonify(
456 {
457 "status": "error",
458 "error_type": "validation",
459 "message": gettext(
460 "Invalid request: 'changes' and"
461 " 'entity_type' are required fields"
462 ),
463 }
464 ),
465 400,
466 )
468 changes = data.get("changes", [])
469 entity_type = data.get("entity_type")
470 entity_shape = data.get("entity_shape")
472 check_for_orphans = orphan_strategy in (
473 OrphanHandlingStrategy.DELETE,
474 OrphanHandlingStrategy.ASK,
475 )
476 check_for_proxies = proxy_strategy in (
477 ProxyHandlingStrategy.DELETE,
478 ProxyHandlingStrategy.ASK,
479 )
481 orphans: list[dict[str, str]] = []
482 intermediate_orphans: list[dict[str, str]] = []
483 if check_for_orphans or check_for_proxies:
484 orphans, intermediate_orphans = _collect_affected_entities(
485 changes,
486 entity_type,
487 check_for_orphans=check_for_orphans,
488 check_for_proxies=check_for_proxies,
489 )
491 if (orphan_strategy == OrphanHandlingStrategy.KEEP or not orphans) and (
492 proxy_strategy == ProxyHandlingStrategy.KEEP or not intermediate_orphans
493 ):
494 return jsonify({"status": "success", "affected_entities": []})
496 return _format_orphan_response(
497 orphans,
498 intermediate_orphans,
499 entity_shape,
500 orphan_strategy,
501 proxy_strategy,
502 )
503 except ValueError as e:
504 error_message = str(e)
505 current_app.logger.warning(
506 "Validation error in check_orphans: %s", error_message
507 )
508 return (
509 jsonify(
510 {
511 "status": "error",
512 "error_type": "validation",
513 "message": gettext(
514 "An error occurred while checking for orphaned entities"
515 ),
516 }
517 ),
518 400,
519 )
520 except Exception as e:
521 error_message = f"Error checking orphans: {e!s}"
522 current_app.logger.exception("%s\n%s", error_message, traceback.format_exc())
523 return (
524 jsonify(
525 {
526 "status": "error",
527 "error_type": "system",
528 "message": gettext(
529 "An error occurred while checking for orphaned entities"
530 ),
531 }
532 ),
533 500,
534 )
537def _parse_change_request(
538 changes: list[dict],
539) -> tuple[URIRef, list[dict], bool, str | None, bool]:
540 first_change = changes[0] if changes else {}
541 subject = URIRef(first_change.get("subject", ""))
542 affected_entities = first_change.get("affected_entities", [])
543 delete_affected = first_change.get("delete_affected", False)
544 primary_source = first_change.get("primary_source")
545 save_default_source = first_change.get("save_default_source", False)
546 return (
547 subject,
548 affected_entities,
549 delete_affected,
550 primary_source,
551 save_default_source,
552 )
555def _affected_entity_will_be_deleted(entity: dict, *, delete_affected: bool) -> bool:
556 orphan_strategy = current_app.config["ORPHAN_HANDLING_STRATEGY"]
557 proxy_strategy = current_app.config["PROXY_HANDLING_STRATEGY"]
559 if entity["is_intermediate"]:
560 return delete_affected and proxy_strategy in (
561 ProxyHandlingStrategy.DELETE,
562 ProxyHandlingStrategy.ASK,
563 )
564 return delete_affected and orphan_strategy in (
565 OrphanHandlingStrategy.DELETE,
566 OrphanHandlingStrategy.ASK,
567 )
570def _collect_entity_deletion_subjects(
571 changes: list[dict],
572 affected_entities: list[dict],
573 *,
574 delete_affected: bool,
575) -> set[URIRef]:
576 deletion_subjects = {
577 URIRef(change["subject"])
578 for change in changes
579 if change["action"] == "delete" and not change.get("predicate")
580 }
582 for entity in affected_entities:
583 if _affected_entity_will_be_deleted(entity, delete_affected=delete_affected):
584 deletion_subjects.add(URIRef(entity["uri"]))
586 return deletion_subjects
589def _setup_editor(
590 primary_source: str | None,
591 changes: list[dict],
592 subject: URIRef,
593 affected_entities: list[dict],
594 *,
595 delete_affected: bool,
596) -> tuple[Editor, URIRef | None]:
597 resp_agent = get_responsible_agent_uri(current_user.orcid)
598 editor = Editor(
599 EndpointConfig(
600 dataset=get_dataset_endpoint(),
601 provenance=get_provenance_endpoint(),
602 is_quadstore=current_app.config["DATASET_IS_QUADSTORE"],
603 ),
604 current_app.config["COUNTER_HANDLER"],
605 resp_agent,
606 URIRef(current_app.config["PRIMARY_SOURCE"]),
607 current_app.config["DATASET_GENERATION_TIME"],
608 save_plugin=current_app.config.get("SAVE_PLUGIN"),
609 )
611 deletion_subjects = _collect_entity_deletion_subjects(
612 changes, affected_entities, delete_affected=delete_affected
613 )
615 editor = import_entity_graph(
616 editor,
617 subject,
618 include_referencing_entities=subject in deletion_subjects,
619 )
620 for deletion_subject in sorted(deletion_subjects - {subject}, key=str):
621 editor = import_entity_graph(
622 editor, deletion_subject, include_referencing_entities=True
623 )
625 for change in changes:
626 if change["action"] == "create":
627 data = change.get("data")
628 if data:
629 import_referenced_entities(editor, data)
631 editor.preexisting_finished()
632 editor.set_primary_source(URIRef(primary_source) if primary_source else None)
634 graph_uri: URIRef | None = None
635 if editor.dataset_is_quadstore:
636 for quad in editor.g_set.quads((subject, None, None, None)): # type: ignore[union-attr]
637 graph_uri = get_graph_uri_from_context(cast("Graph | URIRef", quad[3]))
638 break
640 return editor, graph_uri
643def _process_creates(
644 editor: Editor,
645 changes: list[dict],
646 graph_uri: URIRef | None,
647 subject: URIRef,
648) -> tuple[dict[str, str], URIRef]:
649 temp_id_to_uri: dict[str, str] = {}
650 for change in changes:
651 if change["action"] == "create":
652 data = change.get("data")
653 if data:
654 change_subject_str = change.get("subject")
655 change_subject = (
656 URIRef(change_subject_str) if change_subject_str else None
657 )
658 created_subject = create_logic(
659 editor,
660 data,
661 change_subject,
662 graph_uri,
663 temp_id_to_uri=temp_id_to_uri,
664 parent_entity_type=None,
665 )
666 if change_subject is not None:
667 subject = created_subject
668 return temp_id_to_uri, subject
671def _handle_affected_entities(
672 editor: Editor,
673 affected_entities: list[dict],
674 *,
675 delete_affected: bool,
676 graph_uri: URIRef | None,
677 deleted_entities: set[URIRef],
678) -> None:
679 orphan_strategy = current_app.config.get(
680 "ORPHAN_HANDLING_STRATEGY", OrphanHandlingStrategy.KEEP
681 )
682 proxy_strategy = current_app.config.get(
683 "PROXY_HANDLING_STRATEGY", ProxyHandlingStrategy.KEEP
684 )
685 # Separiamo le operazioni di delete in due fasi:
686 # 1. Prima eliminiamo tutte le entità orfane/intermedie
687 # 2. Poi eliminiamo le triple specifiche
689 # Fase 1: Elimina le entità orfane/intermedie
690 if not (affected_entities and delete_affected):
691 return
693 # Separa gli orfani dalle entità proxy
694 orphans = [
695 entity for entity in affected_entities if not entity.get("is_intermediate")
696 ]
697 proxies = [entity for entity in affected_entities if entity.get("is_intermediate")]
699 # Gestione degli orfani secondo la strategia per gli orfani
700 should_delete_orphans = orphan_strategy == OrphanHandlingStrategy.DELETE or (
701 orphan_strategy == OrphanHandlingStrategy.ASK and delete_affected
702 )
704 if should_delete_orphans and orphans:
705 for orphan in orphans:
706 orphan_uri = URIRef(orphan["uri"])
707 if orphan_uri in deleted_entities:
708 continue
710 delete_logic(
711 ChangeOperation(editor=editor, subject=orphan_uri, graph_uri=graph_uri)
712 )
713 deleted_entities.add(orphan_uri)
715 # Gestione delle entità proxy secondo la strategia per i proxy
716 should_delete_proxies = proxy_strategy == ProxyHandlingStrategy.DELETE or (
717 proxy_strategy == ProxyHandlingStrategy.ASK and delete_affected
718 )
720 if should_delete_proxies and proxies:
721 for proxy in proxies:
722 proxy_uri = URIRef(proxy["uri"])
723 if proxy_uri in deleted_entities:
724 continue
726 delete_logic(
727 ChangeOperation(editor=editor, subject=proxy_uri, graph_uri=graph_uri)
728 )
729 deleted_entities.add(proxy_uri)
732def _process_remaining_changes(
733 editor: Editor,
734 changes: list[dict],
735 graph_uri: URIRef | None,
736 deleted_entities: set[URIRef],
737 temp_id_to_uri: dict[str, str],
738) -> None:
739 for change in changes:
740 if change["action"] == "delete":
741 _process_delete_change(editor, change, graph_uri, deleted_entities)
742 elif change["action"] == "update":
743 op = ChangeOperation(
744 editor=editor,
745 subject=URIRef(change["subject"]),
746 graph_uri=graph_uri,
747 entity_type=change.get("entity_type"),
748 entity_shape=change.get("entity_shape"),
749 )
750 update_logic(
751 op,
752 URIRef(change["predicate"]),
753 change["object"],
754 change["newObject"],
755 )
756 elif change["action"] == "order":
757 op = ChangeOperation(
758 editor=editor,
759 subject=URIRef(change["subject"]),
760 graph_uri=graph_uri,
761 )
762 order_logic(
763 op,
764 URIRef(change["predicate"]),
765 change["object"],
766 URIRef(change["newObject"]),
767 temp_id_to_uri,
768 )
771def _process_delete_change(
772 editor: Editor,
773 change: dict,
774 graph_uri: URIRef | None,
775 deleted_entities: set[URIRef],
776) -> None:
777 change_subject = URIRef(change["subject"])
778 change_predicate = URIRef(change["predicate"]) if change.get("predicate") else None
779 raw_object_value = change.get("object")
780 object_value = str(raw_object_value) if raw_object_value is not None else None
782 op = ChangeOperation(
783 editor=editor,
784 subject=change_subject,
785 graph_uri=graph_uri,
786 entity_type=change.get("entity_type"),
787 entity_shape=change.get("entity_shape"),
788 )
790 if not change_predicate:
791 if change_subject in deleted_entities:
792 return
794 delete_logic(op)
795 deleted_entities.add(change_subject)
796 elif object_value is not None:
797 if is_valid_url(object_value) and URIRef(object_value) in deleted_entities:
798 return
800 delete_logic(op, change_predicate, object_value)
803def _save_and_respond(editor: Editor) -> tuple[Response, int]:
804 try:
805 editor.save()
806 except ValueError:
807 current_app.logger.exception("Error during save operation")
808 raise
809 except Exception as save_error:
810 current_app.logger.exception("Error during save operation")
811 return jsonify(
812 {
813 "status": "error",
814 "error_type": "database",
815 "message": gettext("Failed to save changes to the database: {}").format(
816 str(save_error)
817 ),
818 }
819 ), 500
821 return (
822 jsonify(
823 {
824 "status": "success",
825 "message": gettext("Changes applied successfully"),
826 }
827 ),
828 200,
829 )
832@api_bp.route("/apply_changes", methods=["POST"])
833@login_required
834def apply_changes() -> tuple[Response, int]:
835 """Apply changes to entities.
837 Request body:
838 {
839 "subject": (str) Main entity URI being modified,
840 "changes": (list) List of changes to apply,
841 "primary_source": (str) Primary source to use for provenance,
842 "save_default_source": (bool) Whether to save primary_source as default for
843 current user,
844 "affected_entities": (list) Entities potentially affected by delete operations,
845 "delete_affected": (bool) Whether to delete affected entities
846 }
848 Responses:
849 200 OK: Changes applied successfully
850 400 Bad Request: Invalid request or validation error
851 500 Internal Server Error: Server error while applying changes
852 """
853 try:
854 changes = request.get_json()
855 if not changes:
856 return jsonify({"error": "No request data provided"}), 400
858 (
859 subject,
860 affected_entities,
861 delete_affected,
862 primary_source,
863 save_default_source,
864 ) = _parse_change_request(changes)
866 if primary_source and not is_valid_url(primary_source):
867 return jsonify({"error": "Invalid primary source URL"}), 400
869 if save_default_source and primary_source and is_valid_url(primary_source):
870 save_user_default_primary_source(current_user.orcid, primary_source)
872 changes = transform_changes_with_virtual_properties(changes)
874 editor, graph_uri = _setup_editor(
875 primary_source,
876 changes,
877 subject,
878 affected_entities,
879 delete_affected=delete_affected,
880 )
882 temp_id_to_uri, subject = _process_creates(editor, changes, graph_uri, subject)
884 deleted_entities: set[URIRef] = set()
885 _handle_affected_entities(
886 editor,
887 affected_entities,
888 delete_affected=delete_affected,
889 graph_uri=graph_uri,
890 deleted_entities=deleted_entities,
891 )
892 _process_remaining_changes(
893 editor, changes, graph_uri, deleted_entities, temp_id_to_uri
894 )
896 return _save_and_respond(editor)
898 except ValueError as e:
899 error_message = str(e)
900 current_app.logger.warning("Validation error: %s", error_message)
901 return (
902 jsonify(
903 {
904 "status": "error",
905 "error_type": "validation",
906 "message": error_message,
907 }
908 ),
909 400,
910 )
911 except Exception as e:
912 error_message = f"Error while applying changes: {e!s}\n{traceback.format_exc()}"
913 current_app.logger.exception(error_message)
914 return (
915 jsonify(
916 {
917 "status": "error",
918 "error_type": "system",
919 "message": gettext("An error occurred while applying changes"),
920 }
921 ),
922 500,
923 )
926def get_graph_uri_from_context(graph_context: Graph | URIRef) -> URIRef:
927 if isinstance(graph_context, Graph):
928 return cast("URIRef", graph_context.identifier)
929 return cast("URIRef", graph_context)
932def determine_datatype(value: str, datatype_uris: list[str]) -> URIRef:
933 for datatype_uri in datatype_uris:
934 validation_func = next(
935 (d[1] for d in DATATYPE_MAPPING if str(d[0]) == str(datatype_uri)), None
936 )
937 if validation_func and validation_func(value):
938 return URIRef(datatype_uri)
939 # If none match, default to XSD.string
940 return XSD.string
943class CreateEntityData(TypedDict, total=False):
944 entity_type: str
945 # TODO(arcangelo): tighten this type after normalizing
946 # the frontend payload to a consistent shape
947 properties: dict[str, list | dict | str]
948 tempId: str
951@dataclass
952class _CreateContext:
953 editor: Editor
954 graph_uri: URIRef | None
955 entity_type: str | None
956 temp_id_to_uri: dict[str, str] | None
959def _handle_property_value(
960 ctx: _CreateContext,
961 value: dict | str,
962 subject: URIRef,
963 predicate: URIRef,
964) -> None:
965 if isinstance(value, dict) and "entity_type" in value:
966 nested_subject = generate_unique_uri(value["entity_type"])
967 create_logic(
968 ctx.editor,
969 cast("CreateEntityData", value),
970 nested_subject,
971 ctx.graph_uri,
972 subject,
973 predicate,
974 ctx.temp_id_to_uri,
975 parent_entity_type=ctx.entity_type,
976 )
977 elif isinstance(value, dict) and value.get("is_existing_entity", False):
978 entity_uri = value.get("entity_uri")
979 if entity_uri:
980 ctx.editor.create(subject, predicate, URIRef(entity_uri), ctx.graph_uri)
981 else:
982 msg = "Missing entity_uri in existing entity reference"
983 raise ValueError(msg)
984 elif isinstance(value, dict) and value.get("is_custom_property", False):
985 if value["type"] == "uri":
986 object_value = URIRef(value["value"])
987 elif value["type"] == "literal":
988 datatype = URIRef(value["datatype"]) if "datatype" in value else XSD.string
989 object_value = Literal(value["value"], datatype=datatype)
990 else:
991 msg = f"Unknown custom property type: {value['type']}"
992 raise ValueError(msg)
994 ctx.editor.create(subject, predicate, object_value, ctx.graph_uri)
995 else:
996 object_value, _, error_message = validate_new_triple(
997 subject,
998 predicate,
999 str(value),
1000 "create",
1001 entity_types=ctx.entity_type,
1002 )
1003 if error_message:
1004 raise ValueError(error_message)
1006 if object_value is not None:
1007 ctx.editor.create(subject, predicate, object_value, ctx.graph_uri)
1010def _setup_parent_relations(
1011 ctx: _CreateContext,
1012 subject: URIRef,
1013 parent_subject: URIRef,
1014 parent_predicate: URIRef | None,
1015 parent_entity_type: str | None,
1016) -> None:
1017 type_value, _, error_message = validate_new_triple(
1018 subject, RDF.type, ctx.entity_type, "create", entity_types=ctx.entity_type
1019 )
1020 if error_message:
1021 raise ValueError(error_message)
1023 if type_value is not None:
1024 ctx.editor.create(subject, RDF.type, type_value, ctx.graph_uri)
1026 if parent_predicate:
1027 parent_value, _, error_message = validate_new_triple(
1028 parent_subject,
1029 parent_predicate,
1030 subject,
1031 "create",
1032 entity_types=parent_entity_type,
1033 )
1034 if error_message:
1035 raise ValueError(error_message)
1037 if parent_value is not None:
1038 ctx.editor.create(
1039 parent_subject, parent_predicate, parent_value, ctx.graph_uri
1040 )
1043def create_logic( # noqa: PLR0913
1044 editor: Editor,
1045 data: CreateEntityData,
1046 subject: URIRef | None = None,
1047 graph_uri: URIRef | None = None,
1048 parent_subject: URIRef | None = None,
1049 parent_predicate: URIRef | None = None,
1050 temp_id_to_uri: dict[str, str] | None = None,
1051 parent_entity_type: str | None = None,
1052) -> URIRef:
1053 entity_type: str | None = data.get("entity_type")
1054 properties: dict = data.get("properties", {})
1055 temp_id: str | None = data.get("tempId")
1057 if subject is None:
1058 subject = generate_unique_uri(entity_type, cast("dict", data))
1060 if temp_id and temp_id_to_uri is not None:
1061 temp_id_to_uri[temp_id] = str(subject)
1063 ctx = _CreateContext(editor, graph_uri, entity_type, temp_id_to_uri)
1065 if parent_subject is not None:
1066 _setup_parent_relations(
1067 ctx, subject, parent_subject, parent_predicate, parent_entity_type
1068 )
1070 for predicate_str, values in properties.items():
1071 predicate = URIRef(predicate_str)
1072 values_list = values if isinstance(values, list) else [values]
1073 for value in values_list:
1074 _handle_property_value(ctx, value, subject, predicate)
1076 return subject
1079def update_logic(
1080 op: ChangeOperation,
1081 predicate: URIRef,
1082 old_value: str,
1083 new_value: str,
1084) -> None:
1085 old_value_rdf: URIRef | Literal = (
1086 URIRef(old_value) if is_valid_url(old_value) else Literal(old_value)
1087 )
1088 validated_new, validated_old, error_message = validate_new_triple(
1089 op.subject,
1090 predicate,
1091 new_value,
1092 "update",
1093 old_value_rdf,
1094 entity_types=op.entity_type,
1095 )
1096 if error_message:
1097 raise ValueError(error_message)
1099 op.editor.update(
1100 op.subject,
1101 predicate,
1102 cast("Literal | URIRef", validated_old),
1103 cast("Literal | URIRef", validated_new),
1104 op.graph_uri,
1105 )
1108def rebuild_entity_order(
1109 editor: Editor,
1110 ordered_by_uri: URIRef,
1111 entities: list[URIRef],
1112 graph_uri: URIRef | None = None,
1113) -> Editor:
1114 for entity in entities:
1115 for _s, _p, o in list(
1116 get_triples_from_graph(editor.g_set, (entity, ordered_by_uri, None))
1117 ):
1118 editor.delete(
1119 entity, ordered_by_uri, cast("Literal | URIRef", o), graph_uri
1120 )
1122 # Then rebuild the chain with the entities
1123 for i in range(len(entities) - 1):
1124 current_entity = entities[i]
1125 next_entity = entities[i + 1]
1126 editor.create(current_entity, ordered_by_uri, next_entity, graph_uri)
1128 return editor
1131def delete_logic(
1132 op: ChangeOperation,
1133 predicate: URIRef | None = None,
1134 object_value: str | None = None,
1135) -> None:
1136 resolved_value: URIRef | Literal | None = None
1137 if predicate and object_value:
1138 old_val_rdf: URIRef | Literal = (
1139 URIRef(object_value)
1140 if is_valid_url(object_value)
1141 else Literal(object_value)
1142 )
1143 _, resolved_value, error_message = validate_new_triple(
1144 op.subject,
1145 predicate,
1146 None,
1147 "delete",
1148 old_val_rdf,
1149 entity_types=op.entity_type,
1150 )
1151 if error_message:
1152 raise ValueError(error_message)
1154 op.editor.delete(
1155 op.subject,
1156 predicate,
1157 cast("Literal | URIRef | None", resolved_value),
1158 op.graph_uri,
1159 )
1162def order_logic(
1163 op: ChangeOperation,
1164 predicate: URIRef,
1165 new_order: list[str],
1166 ordered_by: URIRef,
1167 temp_id_to_uri: dict[str, str] | None = None,
1168) -> Editor:
1169 current_entities = [
1170 o
1171 for _, _, o in get_triples_from_graph(
1172 op.editor.g_set, (op.subject, predicate, None)
1173 )
1174 ]
1176 old_to_new_mapping = {}
1178 for old_entity in current_entities:
1179 if str(old_entity) in new_order:
1180 entity_properties = list(
1181 get_triples_from_graph(
1182 op.editor.g_set,
1183 (cast("URIRef", old_entity), None, None),
1184 )
1185 )
1187 entity_type = next(
1188 (o for _, p, o in entity_properties if p == RDF.type), None
1189 )
1191 if entity_type is None:
1192 msg = f"Impossibile determinare il tipo dell'entità per {old_entity}"
1193 raise ValueError(msg)
1195 new_entity_uri = generate_unique_uri(str(entity_type))
1196 old_to_new_mapping[old_entity] = new_entity_uri
1198 op.editor.delete(
1199 op.subject,
1200 predicate,
1201 cast("Literal | URIRef", old_entity),
1202 op.graph_uri,
1203 )
1204 op.editor.delete(cast("URIRef", old_entity), graph=op.graph_uri)
1206 op.editor.create(op.subject, predicate, new_entity_uri, op.graph_uri)
1208 for _, p, o in entity_properties:
1209 if p not in (predicate, ordered_by):
1210 op.editor.create(
1211 new_entity_uri,
1212 cast("URIRef", p),
1213 cast("Literal | URIRef", o),
1214 op.graph_uri,
1215 )
1217 ordered_entities = []
1218 for entity in new_order:
1219 new_entity_uri = old_to_new_mapping.get(URIRef(entity))
1220 if not new_entity_uri:
1221 new_entity_uri = URIRef(
1222 temp_id_to_uri.get(entity, entity) if temp_id_to_uri else entity
1223 )
1224 ordered_entities.append(new_entity_uri)
1226 if ordered_entities:
1227 rebuild_entity_order(op.editor, ordered_by, ordered_entities, op.graph_uri)
1229 return op.editor
1232@api_bp.route("/human-readable-entity", methods=["POST"])
1233@login_required
1234def get_human_readable_entity() -> str | tuple[Response, int]:
1235 custom_filter = get_custom_filter()
1237 # Check if required parameters are present
1238 if "uri" not in request.form or "entity_class" not in request.form:
1239 return jsonify(
1240 {"status": "error", "message": "Missing required parameters"}
1241 ), 400
1243 uri = request.form["uri"]
1244 entity_class = request.form["entity_class"]
1245 shape = determine_shape_for_classes([entity_class])
1246 filter_instance = custom_filter
1247 return filter_instance.human_readable_entity(uri, (entity_class, shape))
1250@api_bp.route("/form-fields", methods=["GET"])
1251@login_required
1252def get_form_fields_for_entity() -> Response | tuple[Response, int]:
1253 """
1254 Get form_fields for a specific entity class and shape combination.
1255 Returns only the requested entity + immediate sub-entities (depth=2) to improve
1256 performance.
1258 Query parameters:
1259 entity_class: URI of the entity class
1260 entity_shape: URI of the entity shape
1262 Returns:
1263 JSON response with form_fields for the specified entity
1264 """
1266 try:
1267 entity_class_decoded = request.args.get("entity_class")
1268 entity_shape_decoded = request.args.get("entity_shape")
1270 if not entity_class_decoded or not entity_shape_decoded:
1271 return jsonify(
1272 {
1273 "status": "error",
1274 "message": (
1275 "Missing required parameters: entity_class and entity_shape"
1276 ),
1277 }
1278 ), 400
1280 all_form_fields = get_form_fields()
1282 if not all_form_fields:
1283 return jsonify(
1284 {"status": "error", "message": "Form fields not initialized"}
1285 ), 500
1287 entity_key = (entity_class_decoded, entity_shape_decoded)
1289 if entity_key not in all_form_fields:
1290 return jsonify(
1291 {
1292 "status": "error",
1293 "message": (
1294 f"No form fields found for entity class"
1295 f" {entity_class_decoded} with shape"
1296 f" {entity_shape_decoded}"
1297 ),
1298 }
1299 ), 404
1301 entity_form_fields = all_form_fields[entity_key]
1303 # Convert OrderedDict to list of [property, details] pairs to preserve order
1304 ordered_properties = []
1305 for prop, details_list in entity_form_fields.items():
1306 ordered_properties.append([prop, details_list])
1308 return jsonify(
1309 {
1310 "status": "success",
1311 "form_fields": ordered_properties,
1312 "entity_key": [entity_class_decoded, entity_shape_decoded],
1313 }
1314 )
1316 except Exception as e:
1317 current_app.logger.exception(
1318 "Error loading form fields for %s/%s",
1319 entity_class_decoded,
1320 entity_shape_decoded,
1321 )
1323 return jsonify(
1324 {"status": "error", "message": f"Failed to load form fields: {e!s}"}
1325 ), 500
1328@api_bp.route("/render-form-fields", methods=["POST"])
1329@login_required
1330def render_form_fields_html() -> str | tuple[Response, int]:
1331 """
1332 Render form fields as HTML for dynamic loading.
1334 Expects JSON payload with:
1335 - entity_key: [entity_class, entity_shape] array
1337 Returns:
1338 HTML string of the rendered form fields
1339 """
1340 try:
1341 data = request.get_json()
1343 if not data or "entity_key" not in data:
1344 return jsonify(
1345 {"status": "error", "message": "Missing required field: entity_key"}
1346 ), 400
1348 entity_key = data["entity_key"] # This is [entity_class, entity_shape] array
1349 entity_class, entity_shape = entity_key
1351 all_form_fields = get_form_fields()
1353 if not all_form_fields:
1354 return jsonify(
1355 {"status": "error", "message": "Form fields not initialized"}
1356 ), 500
1358 tuple_key = (entity_class, entity_shape)
1359 if tuple_key not in all_form_fields:
1360 return jsonify(
1361 {
1362 "status": "error",
1363 "message": (
1364 f"No form fields found for entity"
1365 f" {entity_class} with shape"
1366 f" {entity_shape}"
1367 ),
1368 }
1369 ), 404
1371 entity_form_fields = all_form_fields[tuple_key]
1373 form_fields_array = [
1374 [prop, details_list] for prop, details_list in entity_form_fields.items()
1375 ]
1377 template_string = """
1378 {% from 'macros.jinja' import render_form_field with context %}
1380 {% set entity_type = entity_class %}
1381 {% set entity_shape = entity_shape %}
1382 {% set group_id = ((entity_type, entity_shape) | human_readable_class +
1383 "_group") | replace(" ", "_") %}
1384 <div class="property-group mb-3" id="{{ group_id }}" data-uri="{{ entity_type
1385 }}" data-shape="{{ entity_shape }}">
1386 {% for prop_data in ordered_form_fields %}
1387 {% set prop = prop_data[0] %}
1388 {% set details_list = prop_data[1] %}
1389 {% for details in details_list %}
1390 {{ render_form_field(entity_type, prop, details, all_form_fields) }}
1391 {% endfor %}
1392 {% endfor %}
1393 </div>
1394 """
1396 return render_template_string(
1397 template_string,
1398 entity_class=entity_class,
1399 entity_shape=entity_shape,
1400 ordered_form_fields=form_fields_array,
1401 all_form_fields=all_form_fields,
1402 )
1404 except Exception as e:
1405 current_app.logger.exception("Error rendering form fields HTML")
1407 return jsonify(
1408 {"status": "error", "message": f"Failed to render form fields: {e!s}"}
1409 ), 500
1412def _validate_nested_form_request() -> (
1413 tuple[str, str, str, str, str, int, bool, dict] | tuple[Response, int]
1414):
1415 data = request.get_json()
1417 required_fields = [
1418 "parent_entity_class",
1419 "parent_entity_shape",
1420 "entity_class",
1421 "entity_shape",
1422 "predicate_uri",
1423 "depth",
1424 ]
1426 if not data:
1427 return jsonify({"status": "error", "message": "No JSON data provided"}), 400
1429 missing_fields = [field for field in required_fields if field not in data]
1430 if missing_fields:
1431 return jsonify(
1432 {
1433 "status": "error",
1434 "message": f"Missing required fields: {', '.join(required_fields)}",
1435 }
1436 ), 400
1438 parent_entity_class = data["parent_entity_class"]
1439 parent_entity_shape = data["parent_entity_shape"]
1440 entity_class = data["entity_class"]
1441 entity_shape = data["entity_shape"]
1442 predicate_uri = data["predicate_uri"]
1443 depth = int(data["depth"])
1444 is_template = data.get("is_template", False)
1446 all_form_fields = get_form_fields()
1448 if not all_form_fields:
1449 return jsonify(
1450 {"status": "error", "message": "Form fields not initialized"}
1451 ), 500
1453 parent_entity_key = (parent_entity_class, parent_entity_shape)
1454 if parent_entity_key not in all_form_fields:
1455 return jsonify(
1456 {
1457 "status": "error",
1458 "message": (
1459 "No form fields found for parent"
1460 f" entity {parent_entity_class}"
1461 f" with shape {parent_entity_shape}"
1462 ),
1463 }
1464 ), 404
1466 parent_fields = all_form_fields[parent_entity_key]
1467 if predicate_uri not in parent_fields:
1468 return jsonify(
1469 {
1470 "status": "error",
1471 "message": (
1472 "No field definition found for"
1473 f" predicate {predicate_uri}"
1474 " in parent entity"
1475 ),
1476 }
1477 ), 404
1479 return (
1480 parent_entity_class,
1481 parent_entity_shape,
1482 entity_class,
1483 entity_shape,
1484 predicate_uri,
1485 depth,
1486 is_template,
1487 all_form_fields,
1488 )
1491@api_bp.route("/render-nested-form", methods=["POST"])
1492@login_required
1493def render_nested_form_html() -> str | tuple[Response, int]:
1494 try:
1495 validated = _validate_nested_form_request()
1496 if isinstance(validated[0], Response):
1497 return validated # type: ignore[return-value]
1499 (
1500 parent_entity_class,
1501 parent_entity_shape,
1502 entity_class,
1503 entity_shape,
1504 predicate_uri,
1505 depth,
1506 is_template,
1507 all_form_fields,
1508 ) = cast("tuple[str, str, str, str, str, int, bool, dict]", validated)
1510 parent_entity_key = (parent_entity_class, parent_entity_shape)
1511 parent_fields = all_form_fields[parent_entity_key]
1512 field_details_list = parent_fields[predicate_uri]
1514 target_details = None
1515 for details in field_details_list:
1516 if details.get("or"):
1517 for shape_info in details["or"]:
1518 if (
1519 shape_info.get("entityType") == entity_class
1520 and shape_info.get("nodeShape") == entity_shape
1521 ):
1522 target_details = shape_info
1523 break
1524 if target_details:
1525 break
1527 if not target_details:
1528 return jsonify(
1529 {
1530 "status": "error",
1531 "message": (
1532 "No matching shape info found for"
1533 f" {entity_class}/{entity_shape}"
1534 f" in parent predicate"
1535 f" {predicate_uri}"
1536 ),
1537 }
1538 ), 404
1540 template_string = """
1541 {% from 'macros.jinja' import render_form_field with context %}
1542 {{ render_form_field(parent_entity_class, predicate_uri, shape_info,
1543 all_form_fields, depth, is_template=is_template) }}
1544 """
1546 return render_template_string(
1547 template_string,
1548 parent_entity_class=parent_entity_class,
1549 predicate_uri=predicate_uri,
1550 shape_info=target_details,
1551 all_form_fields=all_form_fields,
1552 depth=depth,
1553 is_template=is_template,
1554 )
1556 except Exception as e:
1557 current_app.logger.exception("Error rendering nested form HTML")
1559 return jsonify(
1560 {"status": "error", "message": f"Failed to render nested form: {e!s}"}
1561 ), 500