Coverage for heritrace/routes/api.py: 99%
559 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-13 22:01 +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.draft_labels import draft_label
33from heritrace.utils.primary_source_utils import save_user_default_primary_source
34from heritrace.utils.shacl_utils import determine_shape_for_classes
35from heritrace.utils.shacl_validation import validate_new_triple
36from heritrace.utils.sparql_utils import (
37 CatalogQuery,
38 DeletedEntitiesQuery,
39 find_orphaned_entities,
40 get_available_classes,
41 get_catalog_data,
42 get_deleted_entities_with_filtering,
43 get_triples_from_graph,
44 import_entity_graph,
45 import_referenced_entities,
46)
47from heritrace.utils.strategies import OrphanHandlingStrategy, ProxyHandlingStrategy
48from heritrace.utils.uri_utils import generate_unique_uri, is_valid_url
49from heritrace.utils.virtual_properties import transform_changes_with_virtual_properties
52@dataclass(frozen=True, slots=True)
53class ChangeOperation:
54 editor: Editor
55 subject: URIRef
56 graph_uri: URIRef | None = None
57 entity_type: str | None = None
58 entity_shape: str | None = None
61api_bp = Blueprint("api", __name__)
64@api_bp.route("/catalogue")
65@login_required
66def catalogue_api() -> Response:
67 selected_class = request.args.get("class")
68 selected_shape = request.args.get("shape")
69 page = int(request.args.get("page", 1))
70 per_page = int(
71 request.args.get("per_page", current_app.config["CATALOGUE_DEFAULT_PER_PAGE"])
72 )
73 sort_property = request.args.get("sort_property")
74 sort_direction = request.args.get("sort_direction", "ASC")
76 allowed_per_page = current_app.config["CATALOGUE_ALLOWED_PER_PAGE"]
77 if per_page not in allowed_per_page:
78 per_page = current_app.config["CATALOGUE_DEFAULT_PER_PAGE"]
80 if not sort_property or sort_property.lower() == "null":
81 sort_property = None
83 available_classes = get_available_classes()
85 catalog_data = get_catalog_data(
86 CatalogQuery(
87 selected_class=selected_class,
88 page=page,
89 per_page=per_page,
90 sort_property=sort_property,
91 sort_direction=sort_direction,
92 selected_shape=selected_shape,
93 ),
94 available_classes,
95 )
97 catalog_data["available_classes"] = available_classes
98 return jsonify(catalog_data)
101@api_bp.route("/time-vault")
102@login_required
103def get_deleted_entities_api() -> Response:
104 """
105 API endpoint to retrieve deleted entities with pagination and sorting.
106 Only processes and returns entities whose classes are marked as visible.
107 """
108 selected_class = request.args.get("class")
109 selected_shape = request.args.get("shape")
110 page = int(request.args.get("page", 1))
111 per_page = int(
112 request.args.get("per_page", current_app.config["CATALOGUE_DEFAULT_PER_PAGE"])
113 )
114 sort_direction = request.args.get("sort_direction", "DESC")
116 allowed_per_page = current_app.config["CATALOGUE_ALLOWED_PER_PAGE"]
117 if per_page not in allowed_per_page:
118 per_page = current_app.config["CATALOGUE_DEFAULT_PER_PAGE"]
120 (
121 deleted_entities,
122 available_classes,
123 selected_class,
124 selected_shape,
125 sortable_properties,
126 total_count,
127 ) = get_deleted_entities_with_filtering(
128 DeletedEntitiesQuery(
129 page,
130 per_page,
131 sort_direction,
132 selected_class,
133 selected_shape,
134 )
135 )
137 return jsonify(
138 {
139 "entities": deleted_entities,
140 "total_pages": (total_count + per_page - 1) // per_page
141 if total_count > 0
142 else 0,
143 "current_page": page,
144 "per_page": per_page,
145 "total_count": total_count,
146 "sort_property": sortable_properties[0]["property"],
147 "sort_direction": sort_direction,
148 "selected_class": selected_class,
149 "selected_shape": selected_shape,
150 "available_classes": available_classes,
151 "sortable_properties": sortable_properties,
152 }
153 )
156@api_bp.route("/check-lock", methods=["POST"])
157@login_required
158def check_lock() -> Response | tuple[Response, int]:
159 """Check if a resource is locked."""
160 try:
161 data = request.get_json()
162 resource_uri = data.get("resource_uri")
164 if not resource_uri:
165 return (
166 jsonify(
167 {"status": "error", "message": gettext("No resource URI provided")}
168 ),
169 400,
170 )
172 status, lock_info = g.resource_lock_manager.check_lock_status(resource_uri)
174 if status == LockStatus.LOCKED:
175 return jsonify(
176 {
177 "status": "locked",
178 "title": gettext("Resource Locked"),
179 "message": gettext(
180 "This resource is currently being"
181 " edited by %(user)s [%(orcid)s]",
182 user=lock_info.user_name,
183 orcid=lock_info.user_id,
184 ),
185 }
186 )
187 if status == LockStatus.ERROR:
188 return (
189 jsonify(
190 {
191 "status": "error",
192 "title": gettext("Error"),
193 "message": gettext("An error occurred while checking the lock"),
194 }
195 ),
196 500,
197 )
198 return jsonify({"status": "available"})
200 except Exception:
201 current_app.logger.exception("Error in check_lock")
202 return (
203 jsonify(
204 {
205 "status": "error",
206 "title": gettext("Error"),
207 "message": gettext("An unexpected error occurred"),
208 }
209 ),
210 500,
211 )
214@api_bp.route("/acquire-lock", methods=["POST"])
215@login_required
216def acquire_lock() -> Response | tuple[Response, int]:
217 """Try to acquire a lock on a resource."""
218 try:
219 data = request.get_json()
220 resource_uri = data.get("resource_uri")
221 linked_resources = data.get("linked_resources", [])
223 if not resource_uri:
224 return (
225 jsonify(
226 {"status": "error", "message": gettext("No resource URI provided")}
227 ),
228 400,
229 )
231 # First check if the resource or any related resource is locked by another user
232 status, lock_info = g.resource_lock_manager.check_lock_status(resource_uri)
233 if status == LockStatus.LOCKED:
234 return (
235 jsonify(
236 {
237 "status": "locked",
238 "title": gettext("Resource Locked"),
239 "message": gettext(
240 "This resource is currently"
241 " being edited by"
242 " %(user)s [%(orcid)s]",
243 user=lock_info.user_name,
244 orcid=lock_info.user_id,
245 ),
246 }
247 ),
248 200,
249 )
251 # Use the provided linked_resources
252 success = g.resource_lock_manager.acquire_lock(resource_uri, linked_resources)
254 if success:
255 return jsonify({"status": "success"})
257 return (
258 jsonify(
259 {
260 "status": "error",
261 "message": gettext("Resource is locked by another user"),
262 }
263 ),
264 423,
265 )
267 except Exception:
268 current_app.logger.exception("Error in acquire_lock")
269 return (
270 jsonify(
271 {"status": "error", "message": gettext("An unexpected error occurred")}
272 ),
273 500,
274 )
277@api_bp.route("/release-lock", methods=["POST"])
278@login_required
279def release_lock() -> Response | tuple[Response, int]:
280 """Release a lock on a resource."""
281 try:
282 data = request.get_json()
283 resource_uri = data.get("resource_uri")
285 if not resource_uri:
286 return (
287 jsonify(
288 {"status": "error", "message": gettext("No resource URI provided")}
289 ),
290 400,
291 )
293 success = g.resource_lock_manager.release_lock(resource_uri)
295 if success:
296 return jsonify({"status": "success"})
298 return (
299 jsonify({"status": "error", "message": gettext("Unable to release lock")}),
300 400,
301 )
303 except Exception:
304 current_app.logger.exception("Error in release_lock")
305 return (
306 jsonify(
307 {"status": "error", "message": gettext("An unexpected error occurred")}
308 ),
309 500,
310 )
313@api_bp.route("/renew-lock", methods=["POST"])
314@login_required
315def renew_lock() -> Response | tuple[Response, int]:
316 """Renew an existing lock on a resource."""
317 try:
318 data = request.get_json()
319 resource_uri = data.get("resource_uri")
321 if not resource_uri:
322 return (
323 jsonify(
324 {"status": "error", "message": gettext("No resource URI provided")}
325 ),
326 400,
327 )
329 # When renewing a lock, we don't need to check for linked resources again
330 # Just pass an empty list as we're only refreshing the existing lock
331 success = g.resource_lock_manager.acquire_lock(resource_uri, [])
333 if success:
334 return jsonify({"status": "success"})
336 return (
337 jsonify({"status": "error", "message": gettext("Unable to renew lock")}),
338 423,
339 )
341 except Exception:
342 current_app.logger.exception("Error in renew_lock")
343 return (
344 jsonify(
345 {"status": "error", "message": gettext("An unexpected error occurred")}
346 ),
347 500,
348 )
351@api_bp.route("/validate-literal", methods=["POST"])
352@login_required
353def validate_literal() -> tuple[Response, int]:
354 """Validate a literal value and suggest appropriate datatypes."""
355 value = request.json.get("value")
356 if not value:
357 return jsonify({"error": gettext("Value is required.")}), 400
359 matching_datatypes = []
360 for datatype, validation_func, _ in DATATYPE_MAPPING:
361 if validation_func(value):
362 matching_datatypes.append(str(datatype))
364 if not matching_datatypes:
365 return jsonify({"error": gettext("No matching datatypes found.")}), 400
367 return jsonify({"valid_datatypes": matching_datatypes}), 200
370def _collect_affected_entities(
371 changes: list[dict],
372 entity_type: str,
373 *,
374 check_for_orphans: bool,
375 check_for_proxies: bool,
376) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
377 orphans: list[dict[str, str]] = []
378 intermediate_orphans: list[dict[str, str]] = []
379 for change in changes:
380 if change["action"] == "delete":
381 found_orphans, found_intermediates = find_orphaned_entities(
382 URIRef(change["subject"]),
383 entity_type,
384 URIRef(change["predicate"]) if change.get("predicate") else None,
385 change.get("object"),
386 )
387 if check_for_orphans:
388 orphans.extend(found_orphans)
389 if check_for_proxies:
390 intermediate_orphans.extend(found_intermediates)
391 return orphans, intermediate_orphans
394def _format_orphan_response(
395 orphans: list[dict[str, str]],
396 intermediate_orphans: list[dict[str, str]],
397 entity_shape: str | None,
398 orphan_strategy: OrphanHandlingStrategy,
399 proxy_strategy: ProxyHandlingStrategy,
400) -> Response:
401 custom_filter = get_custom_filter()
403 def format_entities(
404 entities: list[dict[str, str]],
405 *,
406 is_intermediate: bool = False,
407 ) -> list[dict[str, str | bool]]:
408 return [
409 {
410 "uri": entity["uri"],
411 "label": custom_filter.human_readable_entity(
412 entity["uri"], (entity["type"], entity_shape)
413 ),
414 "type": custom_filter.human_readable_class(
415 (entity["type"], entity_shape)
416 ),
417 "is_intermediate": is_intermediate,
418 }
419 for entity in entities
420 ]
422 affected_entities = format_entities(orphans) + format_entities(
423 intermediate_orphans, is_intermediate=True
424 )
426 should_delete = (
427 orphan_strategy == OrphanHandlingStrategy.DELETE
428 and proxy_strategy == ProxyHandlingStrategy.DELETE
429 )
431 return jsonify(
432 {
433 "status": "success",
434 "affected_entities": affected_entities,
435 "should_delete": should_delete,
436 "orphan_strategy": orphan_strategy.value,
437 "proxy_strategy": proxy_strategy.value,
438 }
439 )
442@api_bp.route("/check_orphans", methods=["POST"])
443@login_required
444def check_orphans() -> Response | tuple[Response, int]:
445 try:
446 orphan_strategy = current_app.config.get(
447 "ORPHAN_HANDLING_STRATEGY", OrphanHandlingStrategy.KEEP
448 )
449 proxy_strategy = current_app.config.get(
450 "PROXY_HANDLING_STRATEGY", ProxyHandlingStrategy.KEEP
451 )
453 data = request.json
454 if not data or "changes" not in data or "entity_type" not in data:
455 return (
456 jsonify(
457 {
458 "status": "error",
459 "error_type": "validation",
460 "message": gettext(
461 "Invalid request: 'changes' and"
462 " 'entity_type' are required fields"
463 ),
464 }
465 ),
466 400,
467 )
469 changes = data.get("changes", [])
470 entity_type = data.get("entity_type")
471 entity_shape = data.get("entity_shape")
473 check_for_orphans = orphan_strategy in (
474 OrphanHandlingStrategy.DELETE,
475 OrphanHandlingStrategy.ASK,
476 )
477 check_for_proxies = proxy_strategy in (
478 ProxyHandlingStrategy.DELETE,
479 ProxyHandlingStrategy.ASK,
480 )
482 orphans: list[dict[str, str]] = []
483 intermediate_orphans: list[dict[str, str]] = []
484 if check_for_orphans or check_for_proxies:
485 orphans, intermediate_orphans = _collect_affected_entities(
486 changes,
487 entity_type,
488 check_for_orphans=check_for_orphans,
489 check_for_proxies=check_for_proxies,
490 )
492 if (orphan_strategy == OrphanHandlingStrategy.KEEP or not orphans) and (
493 proxy_strategy == ProxyHandlingStrategy.KEEP or not intermediate_orphans
494 ):
495 return jsonify({"status": "success", "affected_entities": []})
497 return _format_orphan_response(
498 orphans,
499 intermediate_orphans,
500 entity_shape,
501 orphan_strategy,
502 proxy_strategy,
503 )
504 except ValueError as e:
505 error_message = str(e)
506 current_app.logger.warning(
507 "Validation error in check_orphans: %s", error_message
508 )
509 return (
510 jsonify(
511 {
512 "status": "error",
513 "error_type": "validation",
514 "message": gettext(
515 "An error occurred while checking for orphaned entities"
516 ),
517 }
518 ),
519 400,
520 )
521 except Exception as e:
522 error_message = f"Error checking orphans: {e!s}"
523 current_app.logger.exception("%s\n%s", error_message, traceback.format_exc())
524 return (
525 jsonify(
526 {
527 "status": "error",
528 "error_type": "system",
529 "message": gettext(
530 "An error occurred while checking for orphaned entities"
531 ),
532 }
533 ),
534 500,
535 )
538def _parse_change_request(
539 changes: list[dict],
540) -> tuple[URIRef, list[dict], bool, str | None, bool]:
541 first_change = changes[0] if changes else {}
542 subject = URIRef(first_change.get("subject", ""))
543 affected_entities = first_change.get("affected_entities", [])
544 delete_affected = first_change.get("delete_affected", False)
545 primary_source = first_change.get("primary_source")
546 save_default_source = first_change.get("save_default_source", False)
547 return (
548 subject,
549 affected_entities,
550 delete_affected,
551 primary_source,
552 save_default_source,
553 )
556def _affected_entity_will_be_deleted(entity: dict, *, delete_affected: bool) -> bool:
557 orphan_strategy = current_app.config["ORPHAN_HANDLING_STRATEGY"]
558 proxy_strategy = current_app.config["PROXY_HANDLING_STRATEGY"]
560 if entity["is_intermediate"]:
561 return delete_affected and proxy_strategy in (
562 ProxyHandlingStrategy.DELETE,
563 ProxyHandlingStrategy.ASK,
564 )
565 return delete_affected and orphan_strategy in (
566 OrphanHandlingStrategy.DELETE,
567 OrphanHandlingStrategy.ASK,
568 )
571def _collect_entity_deletion_subjects(
572 changes: list[dict],
573 affected_entities: list[dict],
574 *,
575 delete_affected: bool,
576) -> set[URIRef]:
577 deletion_subjects = {
578 URIRef(change["subject"])
579 for change in changes
580 if change["action"] == "delete" and not change.get("predicate")
581 }
583 for entity in affected_entities:
584 if _affected_entity_will_be_deleted(entity, delete_affected=delete_affected):
585 deletion_subjects.add(URIRef(entity["uri"]))
587 return deletion_subjects
590def _setup_editor(
591 primary_source: str | None,
592 changes: list[dict],
593 subject: URIRef,
594 affected_entities: list[dict],
595 *,
596 delete_affected: bool,
597) -> tuple[Editor, URIRef | None]:
598 resp_agent = get_responsible_agent_uri(current_user.orcid)
599 editor = Editor(
600 EndpointConfig(
601 dataset=get_dataset_endpoint(),
602 provenance=get_provenance_endpoint(),
603 is_quadstore=current_app.config["DATASET_IS_QUADSTORE"],
604 ),
605 current_app.config["COUNTER_HANDLER"],
606 resp_agent,
607 URIRef(current_app.config["PRIMARY_SOURCE"]),
608 current_app.config["DATASET_GENERATION_TIME"],
609 save_plugin=current_app.config.get("SAVE_PLUGIN"),
610 )
612 deletion_subjects = _collect_entity_deletion_subjects(
613 changes, affected_entities, delete_affected=delete_affected
614 )
616 editor = import_entity_graph(
617 editor,
618 subject,
619 include_referencing_entities=subject in deletion_subjects,
620 )
621 for deletion_subject in sorted(deletion_subjects - {subject}, key=str):
622 editor = import_entity_graph(
623 editor, deletion_subject, include_referencing_entities=True
624 )
626 for change in changes:
627 if change["action"] == "create":
628 data = change.get("data")
629 if data:
630 import_referenced_entities(editor, data)
632 editor.preexisting_finished()
633 editor.set_primary_source(URIRef(primary_source) if primary_source else None)
635 graph_uri: URIRef | None = None
636 if editor.dataset_is_quadstore:
637 for quad in editor.g_set.quads((subject, None, None, None)): # type: ignore[union-attr]
638 graph_uri = get_graph_uri_from_context(cast("Graph | URIRef", quad[3]))
639 break
641 return editor, graph_uri
644def _process_creates(
645 editor: Editor,
646 changes: list[dict],
647 graph_uri: URIRef | None,
648 subject: URIRef,
649) -> tuple[dict[str, str], URIRef]:
650 temp_id_to_uri: dict[str, str] = {}
651 for change in changes:
652 if change["action"] == "create":
653 data = change.get("data")
654 if data:
655 change_subject_str = change.get("subject")
656 change_subject = (
657 URIRef(change_subject_str) if change_subject_str else None
658 )
659 created_subject = create_logic(
660 editor,
661 data,
662 change_subject,
663 graph_uri,
664 temp_id_to_uri=temp_id_to_uri,
665 parent_entity_type=None,
666 )
667 if change_subject is not None:
668 subject = created_subject
669 return temp_id_to_uri, subject
672def _handle_affected_entities(
673 editor: Editor,
674 affected_entities: list[dict],
675 *,
676 delete_affected: bool,
677 graph_uri: URIRef | None,
678 deleted_entities: set[URIRef],
679) -> None:
680 orphan_strategy = current_app.config.get(
681 "ORPHAN_HANDLING_STRATEGY", OrphanHandlingStrategy.KEEP
682 )
683 proxy_strategy = current_app.config.get(
684 "PROXY_HANDLING_STRATEGY", ProxyHandlingStrategy.KEEP
685 )
686 # Separiamo le operazioni di delete in due fasi:
687 # 1. Prima eliminiamo tutte le entità orfane/intermedie
688 # 2. Poi eliminiamo le triple specifiche
690 # Fase 1: Elimina le entità orfane/intermedie
691 if not (affected_entities and delete_affected):
692 return
694 # Separa gli orfani dalle entità proxy
695 orphans = [
696 entity for entity in affected_entities if not entity.get("is_intermediate")
697 ]
698 proxies = [entity for entity in affected_entities if entity.get("is_intermediate")]
700 # Gestione degli orfani secondo la strategia per gli orfani
701 should_delete_orphans = orphan_strategy == OrphanHandlingStrategy.DELETE or (
702 orphan_strategy == OrphanHandlingStrategy.ASK and delete_affected
703 )
705 if should_delete_orphans and orphans:
706 for orphan in orphans:
707 orphan_uri = URIRef(orphan["uri"])
708 if orphan_uri in deleted_entities:
709 continue
711 delete_logic(
712 ChangeOperation(editor=editor, subject=orphan_uri, graph_uri=graph_uri)
713 )
714 deleted_entities.add(orphan_uri)
716 # Gestione delle entità proxy secondo la strategia per i proxy
717 should_delete_proxies = proxy_strategy == ProxyHandlingStrategy.DELETE or (
718 proxy_strategy == ProxyHandlingStrategy.ASK and delete_affected
719 )
721 if should_delete_proxies and proxies:
722 for proxy in proxies:
723 proxy_uri = URIRef(proxy["uri"])
724 if proxy_uri in deleted_entities:
725 continue
727 delete_logic(
728 ChangeOperation(editor=editor, subject=proxy_uri, graph_uri=graph_uri)
729 )
730 deleted_entities.add(proxy_uri)
733def _process_remaining_changes(
734 editor: Editor,
735 changes: list[dict],
736 graph_uri: URIRef | None,
737 deleted_entities: set[URIRef],
738 temp_id_to_uri: dict[str, str],
739) -> None:
740 for change in changes:
741 if change["action"] == "delete":
742 _process_delete_change(editor, change, graph_uri, deleted_entities)
743 elif change["action"] == "update":
744 op = ChangeOperation(
745 editor=editor,
746 subject=URIRef(change["subject"]),
747 graph_uri=graph_uri,
748 entity_type=change.get("entity_type"),
749 entity_shape=change.get("entity_shape"),
750 )
751 update_logic(
752 op,
753 URIRef(change["predicate"]),
754 change["object"],
755 change["newObject"],
756 )
757 elif change["action"] == "order":
758 op = ChangeOperation(
759 editor=editor,
760 subject=URIRef(change["subject"]),
761 graph_uri=graph_uri,
762 )
763 order_logic(
764 op,
765 URIRef(change["predicate"]),
766 change["object"],
767 URIRef(change["newObject"]),
768 temp_id_to_uri,
769 )
772def _process_delete_change(
773 editor: Editor,
774 change: dict,
775 graph_uri: URIRef | None,
776 deleted_entities: set[URIRef],
777) -> None:
778 change_subject = URIRef(change["subject"])
779 change_predicate = URIRef(change["predicate"]) if change.get("predicate") else None
780 raw_object_value = change.get("object")
781 object_value = str(raw_object_value) if raw_object_value is not None else None
783 op = ChangeOperation(
784 editor=editor,
785 subject=change_subject,
786 graph_uri=graph_uri,
787 entity_type=change.get("entity_type"),
788 entity_shape=change.get("entity_shape"),
789 )
791 if not change_predicate:
792 if change_subject in deleted_entities:
793 return
795 delete_logic(op)
796 deleted_entities.add(change_subject)
797 elif object_value is not None:
798 if is_valid_url(object_value) and URIRef(object_value) in deleted_entities:
799 return
801 delete_logic(op, change_predicate, object_value)
804def _save_and_respond(editor: Editor) -> tuple[Response, int]:
805 try:
806 editor.save()
807 except ValueError:
808 current_app.logger.exception("Error during save operation")
809 raise
810 except Exception as save_error:
811 current_app.logger.exception("Error during save operation")
812 return jsonify(
813 {
814 "status": "error",
815 "error_type": "database",
816 "message": gettext("Failed to save changes to the database: {}").format(
817 str(save_error)
818 ),
819 }
820 ), 500
822 return (
823 jsonify(
824 {
825 "status": "success",
826 "message": gettext("Changes applied successfully"),
827 }
828 ),
829 200,
830 )
833@api_bp.route("/apply_changes", methods=["POST"])
834@login_required
835def apply_changes() -> tuple[Response, int]:
836 """Apply changes to entities.
838 Request body:
839 {
840 "subject": (str) Main entity URI being modified,
841 "changes": (list) List of changes to apply,
842 "primary_source": (str) Primary source to use for provenance,
843 "save_default_source": (bool) Whether to save primary_source as default for
844 current user,
845 "affected_entities": (list) Entities potentially affected by delete operations,
846 "delete_affected": (bool) Whether to delete affected entities
847 }
849 Responses:
850 200 OK: Changes applied successfully
851 400 Bad Request: Invalid request or validation error
852 500 Internal Server Error: Server error while applying changes
853 """
854 try:
855 changes = request.get_json()
856 if not changes:
857 return jsonify({"error": "No request data provided"}), 400
859 (
860 subject,
861 affected_entities,
862 delete_affected,
863 primary_source,
864 save_default_source,
865 ) = _parse_change_request(changes)
867 if primary_source and not is_valid_url(primary_source):
868 return jsonify({"error": "Invalid primary source URL"}), 400
870 if save_default_source and primary_source and is_valid_url(primary_source):
871 save_user_default_primary_source(current_user.orcid, primary_source)
873 changes = transform_changes_with_virtual_properties(changes)
875 editor, graph_uri = _setup_editor(
876 primary_source,
877 changes,
878 subject,
879 affected_entities,
880 delete_affected=delete_affected,
881 )
883 temp_id_to_uri, subject = _process_creates(editor, changes, graph_uri, subject)
885 deleted_entities: set[URIRef] = set()
886 _handle_affected_entities(
887 editor,
888 affected_entities,
889 delete_affected=delete_affected,
890 graph_uri=graph_uri,
891 deleted_entities=deleted_entities,
892 )
893 _process_remaining_changes(
894 editor, changes, graph_uri, deleted_entities, temp_id_to_uri
895 )
897 return _save_and_respond(editor)
899 except ValueError as e:
900 error_message = str(e)
901 current_app.logger.warning("Validation error: %s", error_message)
902 return (
903 jsonify(
904 {
905 "status": "error",
906 "error_type": "validation",
907 "message": error_message,
908 }
909 ),
910 400,
911 )
912 except Exception as e:
913 error_message = f"Error while applying changes: {e!s}\n{traceback.format_exc()}"
914 current_app.logger.exception(error_message)
915 return (
916 jsonify(
917 {
918 "status": "error",
919 "error_type": "system",
920 "message": gettext("An error occurred while applying changes"),
921 }
922 ),
923 500,
924 )
927def get_graph_uri_from_context(graph_context: Graph | URIRef) -> URIRef:
928 if isinstance(graph_context, Graph):
929 return cast("URIRef", graph_context.identifier)
930 return cast("URIRef", graph_context)
933class CreateEntityData(TypedDict, total=False):
934 entity_type: str
935 # TODO(arcangelo): tighten this type after normalizing
936 # the frontend payload to a consistent shape
937 properties: dict[str, list | dict | str]
938 tempId: str
941@dataclass
942class _CreateContext:
943 editor: Editor
944 graph_uri: URIRef | None
945 entity_type: str | None
946 temp_id_to_uri: dict[str, str] | None
949def _handle_property_value(
950 ctx: _CreateContext,
951 value: dict | str,
952 subject: URIRef,
953 predicate: URIRef,
954) -> None:
955 if isinstance(value, dict) and "entity_type" in value:
956 nested_subject = generate_unique_uri(value["entity_type"])
957 create_logic(
958 ctx.editor,
959 cast("CreateEntityData", value),
960 nested_subject,
961 ctx.graph_uri,
962 subject,
963 predicate,
964 ctx.temp_id_to_uri,
965 parent_entity_type=ctx.entity_type,
966 )
967 elif isinstance(value, dict) and value.get("is_existing_entity", False):
968 entity_uri = value.get("entity_uri")
969 if entity_uri:
970 ctx.editor.create(subject, predicate, URIRef(entity_uri), ctx.graph_uri)
971 else:
972 msg = "Missing entity_uri in existing entity reference"
973 raise ValueError(msg)
974 elif isinstance(value, dict) and value.get("is_custom_property", False):
975 if value["type"] == "uri":
976 object_value = URIRef(value["value"])
977 elif value["type"] == "literal":
978 datatype = URIRef(value["datatype"]) if "datatype" in value else XSD.string
979 object_value = Literal(value["value"], datatype=datatype)
980 else:
981 msg = f"Unknown custom property type: {value['type']}"
982 raise ValueError(msg)
984 ctx.editor.create(subject, predicate, object_value, ctx.graph_uri)
985 else:
986 object_value, _, error_message = validate_new_triple(
987 subject,
988 predicate,
989 str(value),
990 "create",
991 entity_types=ctx.entity_type,
992 )
993 if error_message:
994 raise ValueError(error_message)
996 if object_value is not None:
997 ctx.editor.create(subject, predicate, object_value, ctx.graph_uri)
1000def _setup_parent_relations(
1001 ctx: _CreateContext,
1002 subject: URIRef,
1003 parent_subject: URIRef,
1004 parent_predicate: URIRef | None,
1005 parent_entity_type: str | None,
1006) -> None:
1007 type_value, _, error_message = validate_new_triple(
1008 subject, RDF.type, ctx.entity_type, "create", entity_types=ctx.entity_type
1009 )
1010 if error_message:
1011 raise ValueError(error_message)
1013 if type_value is not None:
1014 ctx.editor.create(subject, RDF.type, type_value, ctx.graph_uri)
1016 if parent_predicate:
1017 parent_value, _, error_message = validate_new_triple(
1018 parent_subject,
1019 parent_predicate,
1020 subject,
1021 "create",
1022 entity_types=parent_entity_type,
1023 )
1024 if error_message:
1025 raise ValueError(error_message)
1027 if parent_value is not None:
1028 ctx.editor.create(
1029 parent_subject, parent_predicate, parent_value, ctx.graph_uri
1030 )
1033def create_logic( # noqa: PLR0913
1034 editor: Editor,
1035 data: CreateEntityData,
1036 subject: URIRef | None = None,
1037 graph_uri: URIRef | None = None,
1038 parent_subject: URIRef | None = None,
1039 parent_predicate: URIRef | None = None,
1040 temp_id_to_uri: dict[str, str] | None = None,
1041 parent_entity_type: str | None = None,
1042) -> URIRef:
1043 entity_type: str | None = data.get("entity_type")
1044 properties: dict = data.get("properties", {})
1045 temp_id: str | None = data.get("tempId")
1047 if subject is None:
1048 subject = generate_unique_uri(entity_type, cast("dict", data))
1050 if temp_id and temp_id_to_uri is not None:
1051 temp_id_to_uri[temp_id] = str(subject)
1053 ctx = _CreateContext(editor, graph_uri, entity_type, temp_id_to_uri)
1055 if parent_subject is not None:
1056 _setup_parent_relations(
1057 ctx, subject, parent_subject, parent_predicate, parent_entity_type
1058 )
1060 for predicate_str, values in properties.items():
1061 predicate = URIRef(predicate_str)
1062 values_list = values if isinstance(values, list) else [values]
1063 for value in values_list:
1064 _handle_property_value(ctx, value, subject, predicate)
1066 return subject
1069def update_logic(
1070 op: ChangeOperation,
1071 predicate: URIRef,
1072 old_value: str,
1073 new_value: str,
1074) -> None:
1075 old_value_rdf: URIRef | Literal = (
1076 URIRef(old_value) if is_valid_url(old_value) else Literal(old_value)
1077 )
1078 validated_new, validated_old, error_message = validate_new_triple(
1079 op.subject,
1080 predicate,
1081 new_value,
1082 "update",
1083 old_value_rdf,
1084 entity_types=op.entity_type,
1085 )
1086 if error_message:
1087 raise ValueError(error_message)
1089 op.editor.update(
1090 op.subject,
1091 predicate,
1092 cast("Literal | URIRef", validated_old),
1093 cast("Literal | URIRef", validated_new),
1094 op.graph_uri,
1095 )
1098def rebuild_entity_order(
1099 editor: Editor,
1100 ordered_by_uri: URIRef,
1101 entities: list[URIRef],
1102 graph_uri: URIRef | None = None,
1103) -> Editor:
1104 for entity in entities:
1105 for _s, _p, o in list(
1106 get_triples_from_graph(editor.g_set, (entity, ordered_by_uri, None))
1107 ):
1108 editor.delete(
1109 entity, ordered_by_uri, cast("Literal | URIRef", o), graph_uri
1110 )
1112 # Then rebuild the chain with the entities
1113 for i in range(len(entities) - 1):
1114 current_entity = entities[i]
1115 next_entity = entities[i + 1]
1116 editor.create(current_entity, ordered_by_uri, next_entity, graph_uri)
1118 return editor
1121def delete_logic(
1122 op: ChangeOperation,
1123 predicate: URIRef | None = None,
1124 object_value: str | None = None,
1125) -> None:
1126 resolved_value: URIRef | Literal | None = None
1127 if predicate and object_value:
1128 old_val_rdf: URIRef | Literal = (
1129 URIRef(object_value)
1130 if is_valid_url(object_value)
1131 else Literal(object_value)
1132 )
1133 _, resolved_value, error_message = validate_new_triple(
1134 op.subject,
1135 predicate,
1136 None,
1137 "delete",
1138 old_val_rdf,
1139 entity_types=op.entity_type,
1140 )
1141 if error_message:
1142 raise ValueError(error_message)
1144 op.editor.delete(
1145 op.subject,
1146 predicate,
1147 cast("Literal | URIRef | None", resolved_value),
1148 op.graph_uri,
1149 )
1152def order_logic(
1153 op: ChangeOperation,
1154 predicate: URIRef,
1155 new_order: list[str],
1156 ordered_by: URIRef,
1157 temp_id_to_uri: dict[str, str] | None = None,
1158) -> Editor:
1159 current_entities = [
1160 o
1161 for _, _, o in get_triples_from_graph(
1162 op.editor.g_set, (op.subject, predicate, None)
1163 )
1164 ]
1166 old_to_new_mapping = {}
1168 for old_entity in current_entities:
1169 if str(old_entity) in new_order:
1170 entity_properties = list(
1171 get_triples_from_graph(
1172 op.editor.g_set,
1173 (cast("URIRef", old_entity), None, None),
1174 )
1175 )
1177 entity_type = next(
1178 (o for _, p, o in entity_properties if p == RDF.type), None
1179 )
1181 if entity_type is None:
1182 msg = f"Impossibile determinare il tipo dell'entità per {old_entity}"
1183 raise ValueError(msg)
1185 new_entity_uri = generate_unique_uri(str(entity_type))
1186 old_to_new_mapping[old_entity] = new_entity_uri
1188 op.editor.delete(
1189 op.subject,
1190 predicate,
1191 cast("Literal | URIRef", old_entity),
1192 op.graph_uri,
1193 )
1194 op.editor.delete(cast("URIRef", old_entity), graph=op.graph_uri)
1196 op.editor.create(op.subject, predicate, new_entity_uri, op.graph_uri)
1198 for _, p, o in entity_properties:
1199 if p not in (predicate, ordered_by):
1200 op.editor.create(
1201 new_entity_uri,
1202 cast("URIRef", p),
1203 cast("Literal | URIRef", o),
1204 op.graph_uri,
1205 )
1207 ordered_entities = []
1208 for entity in new_order:
1209 new_entity_uri = old_to_new_mapping.get(URIRef(entity))
1210 if not new_entity_uri:
1211 new_entity_uri = URIRef(
1212 temp_id_to_uri.get(entity, entity) if temp_id_to_uri else entity
1213 )
1214 ordered_entities.append(new_entity_uri)
1216 if ordered_entities:
1217 rebuild_entity_order(op.editor, ordered_by, ordered_entities, op.graph_uri)
1219 return op.editor
1222@api_bp.route("/human-readable-entity", methods=["POST"])
1223@login_required
1224def get_human_readable_entity() -> str | tuple[Response, int]:
1225 custom_filter = get_custom_filter()
1227 # Check if required parameters are present
1228 if "uri" not in request.form or "entity_class" not in request.form:
1229 return jsonify(
1230 {"status": "error", "message": "Missing required parameters"}
1231 ), 400
1233 uri = request.form["uri"]
1234 entity_class = request.form["entity_class"]
1235 shape = determine_shape_for_classes([entity_class])
1236 filter_instance = custom_filter
1237 return filter_instance.human_readable_entity(uri, (entity_class, shape))
1240@api_bp.route("/form-fields", methods=["GET"])
1241@login_required
1242def get_form_fields_for_entity() -> Response | tuple[Response, int]:
1243 """
1244 Get form_fields for a specific entity class and shape combination.
1245 Returns only the requested entity + immediate sub-entities (depth=2) to improve
1246 performance.
1248 Query parameters:
1249 entity_class: URI of the entity class
1250 entity_shape: URI of the entity shape
1252 Returns:
1253 JSON response with form_fields for the specified entity
1254 """
1256 try:
1257 entity_class_decoded = request.args.get("entity_class")
1258 entity_shape_decoded = request.args.get("entity_shape")
1260 if not entity_class_decoded or not entity_shape_decoded:
1261 return jsonify(
1262 {
1263 "status": "error",
1264 "message": (
1265 "Missing required parameters: entity_class and entity_shape"
1266 ),
1267 }
1268 ), 400
1270 all_form_fields = get_form_fields()
1272 if not all_form_fields:
1273 return jsonify(
1274 {"status": "error", "message": "Form fields not initialized"}
1275 ), 500
1277 entity_key = (entity_class_decoded, entity_shape_decoded)
1279 if entity_key not in all_form_fields:
1280 return jsonify(
1281 {
1282 "status": "error",
1283 "message": (
1284 f"No form fields found for entity class"
1285 f" {entity_class_decoded} with shape"
1286 f" {entity_shape_decoded}"
1287 ),
1288 }
1289 ), 404
1291 entity_form_fields = all_form_fields[entity_key]
1293 # Convert OrderedDict to list of [property, details] pairs to preserve order
1294 ordered_properties = []
1295 for prop, details_list in entity_form_fields.items():
1296 ordered_properties.append([prop, details_list])
1298 return jsonify(
1299 {
1300 "status": "success",
1301 "form_fields": ordered_properties,
1302 "entity_key": [entity_class_decoded, entity_shape_decoded],
1303 }
1304 )
1306 except Exception as e:
1307 current_app.logger.exception(
1308 "Error loading form fields for %s/%s",
1309 entity_class_decoded,
1310 entity_shape_decoded,
1311 )
1313 return jsonify(
1314 {"status": "error", "message": f"Failed to load form fields: {e!s}"}
1315 ), 500
1318@api_bp.route("/render-form-fields", methods=["POST"])
1319@login_required
1320def render_form_fields_html() -> str | tuple[Response, int]:
1321 """
1322 Render form fields as HTML for dynamic loading.
1324 Expects JSON payload with:
1325 - entity_key: [entity_class, entity_shape] array
1327 Returns:
1328 HTML string of the rendered form fields
1329 """
1330 try:
1331 data = request.get_json()
1333 if not data or "entity_key" not in data:
1334 return jsonify(
1335 {"status": "error", "message": "Missing required field: entity_key"}
1336 ), 400
1338 entity_key = data["entity_key"] # This is [entity_class, entity_shape] array
1339 entity_class, entity_shape = entity_key
1341 all_form_fields = get_form_fields()
1343 if not all_form_fields:
1344 return jsonify(
1345 {"status": "error", "message": "Form fields not initialized"}
1346 ), 500
1348 tuple_key = (entity_class, entity_shape)
1349 if tuple_key not in all_form_fields:
1350 return jsonify(
1351 {
1352 "status": "error",
1353 "message": (
1354 f"No form fields found for entity"
1355 f" {entity_class} with shape"
1356 f" {entity_shape}"
1357 ),
1358 }
1359 ), 404
1361 entity_form_fields = all_form_fields[tuple_key]
1363 form_fields_array = [
1364 [prop, details_list] for prop, details_list in entity_form_fields.items()
1365 ]
1367 template_string = """
1368 {% from 'macros.jinja' import render_form_field with context %}
1370 {% set entity_type = entity_class %}
1371 {% set entity_shape = entity_shape %}
1372 {% set group_id = ((entity_type, entity_shape) | human_readable_class +
1373 "_group") | replace(" ", "_") %}
1374 <div class="property-group mb-3" id="{{ group_id }}" data-uri="{{ entity_type
1375 }}" data-shape="{{ entity_shape }}">
1376 {% for prop_data in ordered_form_fields %}
1377 {% set prop = prop_data[0] %}
1378 {% set details_list = prop_data[1] %}
1379 {% for details in details_list %}
1380 {{ render_form_field(entity_type, prop, details, all_form_fields) }}
1381 {% endfor %}
1382 {% endfor %}
1383 </div>
1384 """
1386 return render_template_string(
1387 template_string,
1388 entity_class=entity_class,
1389 entity_shape=entity_shape,
1390 ordered_form_fields=form_fields_array,
1391 all_form_fields=all_form_fields,
1392 )
1394 except Exception as e:
1395 current_app.logger.exception("Error rendering form fields HTML")
1397 return jsonify(
1398 {"status": "error", "message": f"Failed to render form fields: {e!s}"}
1399 ), 500
1402def _validate_nested_form_request() -> (
1403 tuple[str, str, str, str, str, int, bool, dict] | tuple[Response, int]
1404):
1405 data = request.get_json()
1407 required_fields = [
1408 "parent_entity_class",
1409 "parent_entity_shape",
1410 "entity_class",
1411 "entity_shape",
1412 "predicate_uri",
1413 "depth",
1414 ]
1416 if not data:
1417 return jsonify({"status": "error", "message": "No JSON data provided"}), 400
1419 missing_fields = [field for field in required_fields if field not in data]
1420 if missing_fields:
1421 return jsonify(
1422 {
1423 "status": "error",
1424 "message": f"Missing required fields: {', '.join(required_fields)}",
1425 }
1426 ), 400
1428 parent_entity_class = data["parent_entity_class"]
1429 parent_entity_shape = data["parent_entity_shape"]
1430 entity_class = data["entity_class"]
1431 entity_shape = data["entity_shape"]
1432 predicate_uri = data["predicate_uri"]
1433 depth = int(data["depth"])
1434 is_template = data.get("is_template", False)
1436 all_form_fields = get_form_fields()
1438 if not all_form_fields:
1439 return jsonify(
1440 {"status": "error", "message": "Form fields not initialized"}
1441 ), 500
1443 parent_entity_key = (parent_entity_class, parent_entity_shape)
1444 if parent_entity_key not in all_form_fields:
1445 return jsonify(
1446 {
1447 "status": "error",
1448 "message": (
1449 "No form fields found for parent"
1450 f" entity {parent_entity_class}"
1451 f" with shape {parent_entity_shape}"
1452 ),
1453 }
1454 ), 404
1456 parent_fields = all_form_fields[parent_entity_key]
1457 if predicate_uri not in parent_fields:
1458 return jsonify(
1459 {
1460 "status": "error",
1461 "message": (
1462 "No field definition found for"
1463 f" predicate {predicate_uri}"
1464 " in parent entity"
1465 ),
1466 }
1467 ), 404
1469 return (
1470 parent_entity_class,
1471 parent_entity_shape,
1472 entity_class,
1473 entity_shape,
1474 predicate_uri,
1475 depth,
1476 is_template,
1477 all_form_fields,
1478 )
1481@api_bp.route("/render-nested-form", methods=["POST"])
1482@login_required
1483def render_nested_form_html() -> str | tuple[Response, int]:
1484 try:
1485 validated = _validate_nested_form_request()
1486 if isinstance(validated[0], Response):
1487 return validated # type: ignore[return-value]
1489 (
1490 parent_entity_class,
1491 parent_entity_shape,
1492 entity_class,
1493 entity_shape,
1494 predicate_uri,
1495 depth,
1496 is_template,
1497 all_form_fields,
1498 ) = cast("tuple[str, str, str, str, str, int, bool, dict]", validated)
1500 parent_entity_key = (parent_entity_class, parent_entity_shape)
1501 parent_fields = all_form_fields[parent_entity_key]
1502 field_details_list = parent_fields[predicate_uri]
1504 target_details = None
1505 for details in field_details_list:
1506 if details.get("or"):
1507 for shape_info in details["or"]:
1508 if (
1509 shape_info.get("entityType") == entity_class
1510 and shape_info.get("nodeShape") == entity_shape
1511 ):
1512 target_details = shape_info
1513 break
1514 if target_details:
1515 break
1517 if not target_details:
1518 return jsonify(
1519 {
1520 "status": "error",
1521 "message": (
1522 "No matching shape info found for"
1523 f" {entity_class}/{entity_shape}"
1524 f" in parent predicate"
1525 f" {predicate_uri}"
1526 ),
1527 }
1528 ), 404
1530 template_string = """
1531 {% from 'macros.jinja' import render_form_field with context %}
1532 {{ render_form_field(parent_entity_class, predicate_uri, shape_info,
1533 all_form_fields, depth, is_template=is_template) }}
1534 """
1536 return render_template_string(
1537 template_string,
1538 parent_entity_class=parent_entity_class,
1539 predicate_uri=predicate_uri,
1540 shape_info=target_details,
1541 all_form_fields=all_form_fields,
1542 depth=depth,
1543 is_template=is_template,
1544 )
1546 except Exception as e:
1547 current_app.logger.exception("Error rendering nested form HTML")
1549 return jsonify(
1550 {"status": "error", "message": f"Failed to render nested form: {e!s}"}
1551 ), 500
1554@api_bp.route("/draft-labels", methods=["POST"])
1555@login_required
1556def preview_draft_labels() -> Response | tuple[Response, int]:
1557 data = request.get_json()
1558 if not isinstance(data, dict) or not isinstance(data.get("entries"), list):
1559 return jsonify({"error": "Expected a list of draft entries"}), 400
1560 for entry in data["entries"]:
1561 if not isinstance(entry, dict) or not all(
1562 key in entry
1563 for key in ("parent_class", "parent_shape", "predicate", "entity")
1564 ):
1565 return jsonify({"error": "Invalid draft entry"}), 400
1566 entity = entry["entity"]
1567 if not isinstance(entity, dict) or not all(
1568 key in entity for key in ("entity_type", "entity_shape", "properties")
1569 ):
1570 return jsonify({"error": "Invalid draft entity"}), 400
1571 return jsonify({"labels": [draft_label(entry) for entry in data["entries"]]})