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

1# heritrace/routes/api.py 

2 

3import traceback 

4from typing import Dict, Optional 

5 

6from flask import Blueprint, current_app, g, jsonify, request 

7from flask_babel import gettext 

8from flask_login import current_user, login_required 

9from heritrace.editor import Editor 

10from heritrace.extensions import (get_custom_filter, get_dataset_endpoint, 

11 get_provenance_endpoint) 

12from heritrace.services.resource_lock_manager import LockStatus 

13from heritrace.utils.shacl_utils import validate_new_triple 

14from heritrace.utils.sparql_utils import (find_orphaned_entities, 

15 get_available_classes, 

16 get_catalog_data, 

17 get_deleted_entities_with_filtering, 

18 import_entity_graph) 

19from heritrace.utils.strategies import (OrphanHandlingStrategy, 

20 ProxyHandlingStrategy) 

21from heritrace.utils.uri_utils import generate_unique_uri 

22from rdflib import RDF, XSD, Graph, URIRef 

23from resources.datatypes import DATATYPE_MAPPING 

24from SPARQLWrapper import JSON 

25 

26api_bp = Blueprint("api", __name__) 

27 

28 

29@api_bp.route("/catalogue") 

30@login_required 

31def catalogue_api(): 

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

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

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

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

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

37 

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

39 if per_page not in allowed_per_page: 

40 per_page = 100 

41 

42 if not sort_property or sort_property.lower() == "null": 

43 sort_property = None 

44 

45 catalog_data = get_catalog_data( 

46 selected_class, page, per_page, sort_property, sort_direction 

47 ) 

48 

49 catalog_data["available_classes"] = get_available_classes() 

50 return jsonify(catalog_data) 

51 

52 

53@api_bp.route("/time-vault") 

54@login_required 

55def get_deleted_entities_api(): 

56 """ 

57 API endpoint to retrieve deleted entities with pagination and sorting. 

58 Only processes and returns entities whose classes are marked as visible. 

59 """ 

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

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

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

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

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

65 

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

67 if per_page not in allowed_per_page: 

68 per_page = 100 

69 

70 deleted_entities, available_classes, selected_class, sortable_properties, total_count = ( 

71 get_deleted_entities_with_filtering( 

72 page, per_page, sort_property, sort_direction, selected_class 

73 ) 

74 ) 

75 

76 return jsonify( 

77 { 

78 "entities": deleted_entities, 

79 "total_pages": (total_count + per_page - 1) // per_page if total_count > 0 else 0, 

80 "current_page": page, 

81 "per_page": per_page, 

82 "total_count": total_count, 

83 "sort_property": sort_property, 

84 "sort_direction": sort_direction, 

85 "selected_class": selected_class, 

86 "available_classes": available_classes, 

87 "sortable_properties": sortable_properties, 

88 } 

89 ) 

90 

91 

92@api_bp.route("/check-lock", methods=["POST"]) 

93@login_required 

94def check_lock(): 

95 """Check if a resource is locked.""" 

96 try: 

97 data = request.get_json() 

98 resource_uri = data.get("resource_uri") 

99 

100 if not resource_uri: 

101 return ( 

102 jsonify( 

103 {"status": "error", "message": gettext("No resource URI provided")} 

104 ), 

105 400, 

106 ) 

107 

108 status, lock_info = g.resource_lock_manager.check_lock_status(resource_uri) 

109 

110 if status == LockStatus.LOCKED: 

111 return jsonify( 

112 { 

113 "status": "locked", 

114 "title": gettext("Resource Locked"), 

115 "message": gettext( 

116 "This resource is currently being edited by %(user)s [%(orcid)s]", 

117 user=lock_info.user_name, 

118 orcid=lock_info.user_id, 

119 ), 

120 } 

121 ) 

122 elif status == LockStatus.ERROR: 

123 return ( 

124 jsonify( 

125 { 

126 "status": "error", 

127 "title": gettext("Error"), 

128 "message": gettext("An error occurred while checking the lock"), 

129 } 

130 ), 

131 500, 

132 ) 

133 else: 

134 return jsonify({"status": "available"}) 

135 

136 except Exception as e: 

137 current_app.logger.error(f"Error in check_lock: {str(e)}") 

138 return ( 

139 jsonify( 

140 { 

141 "status": "error", 

142 "title": gettext("Error"), 

143 "message": gettext("An unexpected error occurred"), 

144 } 

145 ), 

146 500, 

147 ) 

148 

149 

150@api_bp.route("/acquire-lock", methods=["POST"]) 

151@login_required 

152def acquire_lock(): 

153 """Try to acquire a lock on a resource.""" 

154 try: 

155 data = request.get_json() 

156 resource_uri = data.get("resource_uri") 

157 linked_resources = data.get("linked_resources", []) 

158 

159 if not resource_uri: 

160 return ( 

161 jsonify( 

162 {"status": "error", "message": gettext("No resource URI provided")} 

163 ), 

164 400, 

165 ) 

166 

167 # First check if the resource or any related resource is locked by another user 

168 status, lock_info = g.resource_lock_manager.check_lock_status(resource_uri) 

169 if status == LockStatus.LOCKED: 

170 return ( 

171 jsonify( 

172 { 

173 "status": "locked", 

174 "title": gettext("Resource Locked"), 

175 "message": gettext( 

176 "This resource is currently being edited by %(user)s [%(orcid)s]", 

177 user=lock_info.user_name, 

178 orcid=lock_info.user_id, 

179 ), 

180 } 

181 ), 

182 200, 

183 ) 

184 

185 # Use the provided linked_resources 

186 success = g.resource_lock_manager.acquire_lock(resource_uri, linked_resources) 

187 

188 if success: 

189 return jsonify({"status": "success"}) 

190 

191 return ( 

192 jsonify( 

193 { 

194 "status": "error", 

195 "message": gettext("Resource is locked by another user"), 

196 } 

197 ), 

198 423, 

199 ) 

200 

201 except Exception as e: 

202 current_app.logger.error(f"Error in acquire_lock: {str(e)}") 

203 return ( 

204 jsonify( 

205 {"status": "error", "message": gettext("An unexpected error occurred")} 

206 ), 

207 500, 

208 ) 

209 

210 

211@api_bp.route("/release-lock", methods=["POST"]) 

212@login_required 

213def release_lock(): 

214 """Release a lock on a resource.""" 

215 try: 

216 data = request.get_json() 

217 resource_uri = data.get("resource_uri") 

218 

219 if not resource_uri: 

220 return ( 

221 jsonify( 

222 {"status": "error", "message": gettext("No resource URI provided")} 

223 ), 

224 400, 

225 ) 

226 

227 success = g.resource_lock_manager.release_lock(resource_uri) 

228 

229 if success: 

230 return jsonify({"status": "success"}) 

231 

232 return ( 

233 jsonify({"status": "error", "message": gettext("Unable to release lock")}), 

234 400, 

235 ) 

236 

237 except Exception as e: 

238 current_app.logger.error(f"Error in release_lock: {str(e)}") 

239 return ( 

240 jsonify( 

241 {"status": "error", "message": gettext("An unexpected error occurred")} 

242 ), 

243 500, 

244 ) 

245 

246 

247@api_bp.route("/renew-lock", methods=["POST"]) 

248@login_required 

249def renew_lock(): 

250 """Renew an existing lock on a resource.""" 

251 try: 

252 data = request.get_json() 

253 resource_uri = data.get("resource_uri") 

254 

255 if not resource_uri: 

256 return ( 

257 jsonify( 

258 {"status": "error", "message": gettext("No resource URI provided")} 

259 ), 

260 400, 

261 ) 

262 

263 # When renewing a lock, we don't need to check for linked resources again 

264 # Just pass an empty list as we're only refreshing the existing lock 

265 success = g.resource_lock_manager.acquire_lock(resource_uri, []) 

266 

267 if success: 

268 return jsonify({"status": "success"}) 

269 

270 return ( 

271 jsonify({"status": "error", "message": gettext("Unable to renew lock")}), 

272 423, 

273 ) 

274 

275 except Exception as e: 

276 current_app.logger.error(f"Error in renew_lock: {str(e)}") 

277 return ( 

278 jsonify( 

279 {"status": "error", "message": gettext("An unexpected error occurred")} 

280 ), 

281 500, 

282 ) 

283 

284 

285@api_bp.route("/validate-literal", methods=["POST"]) 

286@login_required 

287def validate_literal(): 

288 """Validate a literal value and suggest appropriate datatypes.""" 

289 value = request.json.get("value") 

290 if not value: 

291 return jsonify({"error": gettext("Value is required.")}), 400 

292 

293 matching_datatypes = [] 

294 for datatype, validation_func, _ in DATATYPE_MAPPING: 

295 if validation_func(value): 

296 matching_datatypes.append(str(datatype)) 

297 

298 if not matching_datatypes: 

299 return jsonify({"error": gettext("No matching datatypes found.")}), 400 

300 

301 return jsonify({"valid_datatypes": matching_datatypes}), 200 

302 

303 

304@api_bp.route("/check_orphans", methods=["POST"]) 

305@login_required 

306def check_orphans(): 

307 """ 

308 Check for orphaned entities and intermediate relations (proxies) that would result from the requested changes. 

309 Applies separate handling strategies for orphans and proxies, but returns a unified report. 

310 """ 

311 try: 

312 # Get strategies from configuration 

313 orphan_strategy = current_app.config.get( 

314 "ORPHAN_HANDLING_STRATEGY", OrphanHandlingStrategy.KEEP 

315 ) 

316 proxy_strategy = current_app.config.get( 

317 "PROXY_HANDLING_STRATEGY", ProxyHandlingStrategy.KEEP 

318 ) 

319 

320 data = request.json 

321 # Validate required fields 

322 if not data or "changes" not in data or "entity_type" not in data: 

323 return ( 

324 jsonify( 

325 { 

326 "status": "error", 

327 "error_type": "validation", 

328 "message": gettext( 

329 "Invalid request: 'changes' and 'entity_type' are required fields" 

330 ), 

331 } 

332 ), 

333 400, 

334 ) 

335 

336 changes = data.get("changes", []) 

337 entity_type = data.get("entity_type") 

338 custom_filter = get_custom_filter() 

339 

340 orphans = [] 

341 intermediate_orphans = [] 

342 

343 # Check for orphans and proxies based on their respective strategies 

344 check_for_orphans = orphan_strategy in ( 

345 OrphanHandlingStrategy.DELETE, 

346 OrphanHandlingStrategy.ASK, 

347 ) 

348 check_for_proxies = proxy_strategy in ( 

349 ProxyHandlingStrategy.DELETE, 

350 ProxyHandlingStrategy.ASK, 

351 ) 

352 if check_for_orphans or check_for_proxies: 

353 for change in changes: 

354 if change["action"] == "delete": 

355 found_orphans, found_intermediates = find_orphaned_entities( 

356 change["subject"], 

357 entity_type, 

358 change.get("predicate"), 

359 change.get("object"), 

360 ) 

361 # Only collect orphans if we need to handle them 

362 if check_for_orphans: 

363 orphans.extend(found_orphans) 

364 

365 # Only collect proxies if we need to handle them 

366 if check_for_proxies: 

367 intermediate_orphans.extend(found_intermediates) 

368 

369 # If both strategies are KEEP or no entities found, return empty result 

370 if (orphan_strategy == OrphanHandlingStrategy.KEEP or not orphans) and ( 

371 proxy_strategy == ProxyHandlingStrategy.KEEP or not intermediate_orphans 

372 ): 

373 return jsonify({"status": "success", "affected_entities": []}) 

374 

375 # Format entities for display 

376 def format_entities(entities, is_intermediate=False): 

377 return [ 

378 { 

379 "uri": entity["uri"], 

380 "label": custom_filter.human_readable_entity( 

381 entity["uri"], [entity["type"]] 

382 ), 

383 "type": custom_filter.human_readable_predicate( 

384 entity["type"], [entity["type"]] 

385 ), 

386 "is_intermediate": is_intermediate, 

387 } 

388 for entity in entities 

389 ] 

390 

391 # Create a unified list of affected entities 

392 affected_entities = format_entities(orphans) + format_entities( 

393 intermediate_orphans, is_intermediate=True 

394 ) 

395 

396 # Determine if we should automatically delete entities 

397 should_delete_orphans = orphan_strategy == OrphanHandlingStrategy.DELETE 

398 should_delete_proxies = proxy_strategy == ProxyHandlingStrategy.DELETE 

399 

400 # If both strategies are DELETE, we can automatically delete everything 

401 if should_delete_orphans and should_delete_proxies: 

402 return jsonify( 

403 { 

404 "status": "success", 

405 "affected_entities": affected_entities, 

406 "should_delete": True, 

407 "orphan_strategy": orphan_strategy.value, 

408 "proxy_strategy": proxy_strategy.value, 

409 } 

410 ) 

411 

412 # If at least one strategy is ASK, we need to ask the user 

413 return jsonify( 

414 { 

415 "status": "success", 

416 "affected_entities": affected_entities, 

417 "should_delete": False, 

418 "orphan_strategy": orphan_strategy.value, 

419 "proxy_strategy": proxy_strategy.value, 

420 } 

421 ) 

422 except ValueError as e: 

423 # Handle validation errors specifically 

424 error_message = str(e) 

425 current_app.logger.warning( 

426 f"Validation error in check_orphans: {error_message}" 

427 ) 

428 return ( 

429 jsonify( 

430 { 

431 "status": "error", 

432 "error_type": "validation", 

433 "message": gettext( 

434 "An error occurred while checking for orphaned entities" 

435 ), 

436 } 

437 ), 

438 400, 

439 ) 

440 except Exception as e: 

441 # Handle other errors 

442 error_message = f"Error checking orphans: {str(e)}" 

443 current_app.logger.error(f"{error_message}\n{traceback.format_exc()}") 

444 return ( 

445 jsonify( 

446 { 

447 "status": "error", 

448 "error_type": "system", 

449 "message": gettext( 

450 "An error occurred while checking for orphaned entities" 

451 ), 

452 } 

453 ), 

454 500, 

455 ) 

456 

457 

458@api_bp.route("/apply_changes", methods=["POST"]) 

459@login_required 

460def apply_changes(): 

461 try: 

462 # Remove all debug logging statements 

463 changes = request.json 

464 subject = changes[0]["subject"] 

465 affected_entities = changes[0].get("affected_entities", []) 

466 delete_affected = changes[0].get("delete_affected", False) 

467 

468 # Tieni traccia delle entità già eliminate per evitare duplicazioni 

469 deleted_entities = set() 

470 editor = Editor( 

471 get_dataset_endpoint(), 

472 get_provenance_endpoint(), 

473 current_app.config["COUNTER_HANDLER"], 

474 URIRef(f"https://orcid.org/{current_user.orcid}"), 

475 current_app.config["PRIMARY_SOURCE"], 

476 current_app.config["DATASET_GENERATION_TIME"], 

477 dataset_is_quadstore=current_app.config["DATASET_IS_QUADSTORE"], 

478 ) 

479 

480 # Se c'è un'operazione di eliminazione completa dell'entità, includiamo le entità referenzianti 

481 has_entity_deletion = any( 

482 change["action"] == "delete" and not change.get("predicate") 

483 for change in changes 

484 ) 

485 

486 # Import entity con l'opzione per includere le entità referenzianti se necessario 

487 editor = import_entity_graph( 

488 editor, 

489 subject, 

490 include_referencing_entities=has_entity_deletion 

491 ) 

492 editor.preexisting_finished() 

493 

494 graph_uri = None 

495 if editor.dataset_is_quadstore: 

496 for quad in editor.g_set.quads((URIRef(subject), None, None, None)): 

497 # Ottieni direttamente l'identificatore del grafo 

498 graph_context = quad[3] 

499 graph_uri = get_graph_uri_from_context(graph_context) 

500 break 

501 

502 # Gestisci prima le creazioni 

503 temp_id_to_uri = {} 

504 for change in changes: 

505 if change["action"] == "create": 

506 data = change.get("data") 

507 if data: 

508 subject = create_logic( 

509 editor, 

510 data, 

511 subject, 

512 graph_uri, 

513 temp_id_to_uri=temp_id_to_uri, 

514 parent_entity_type=None, 

515 ) 

516 

517 # Poi gestisci le altre modifiche 

518 orphan_strategy = current_app.config.get( 

519 "ORPHAN_HANDLING_STRATEGY", OrphanHandlingStrategy.KEEP 

520 ) 

521 proxy_strategy = current_app.config.get( 

522 "PROXY_HANDLING_STRATEGY", ProxyHandlingStrategy.KEEP 

523 ) 

524 # Separiamo le operazioni di delete in due fasi: 

525 # 1. Prima eliminiamo tutte le entità orfane/intermedie 

526 # 2. Poi eliminiamo le triple specifiche 

527 

528 # Fase 1: Elimina le entità orfane/intermedie 

529 if affected_entities and delete_affected: 

530 # Separa gli orfani dalle entità proxy 

531 orphans = [entity for entity in affected_entities if not entity.get("is_intermediate")] 

532 proxies = [entity for entity in affected_entities if entity.get("is_intermediate")] 

533 

534 # Gestione degli orfani secondo la strategia per gli orfani 

535 should_delete_orphans = ( 

536 orphan_strategy == OrphanHandlingStrategy.DELETE 

537 or (orphan_strategy == OrphanHandlingStrategy.ASK and delete_affected) 

538 ) 

539 

540 if should_delete_orphans and orphans: 

541 for orphan in orphans: 

542 orphan_uri = orphan["uri"] 

543 if orphan_uri in deleted_entities: 

544 continue 

545 

546 delete_logic(editor, orphan_uri, graph_uri=graph_uri) 

547 deleted_entities.add(orphan_uri) 

548 

549 # Gestione delle entità proxy secondo la strategia per i proxy 

550 should_delete_proxies = ( 

551 proxy_strategy == ProxyHandlingStrategy.DELETE 

552 or (proxy_strategy == ProxyHandlingStrategy.ASK and delete_affected) 

553 ) 

554 

555 if should_delete_proxies and proxies: 

556 for proxy in proxies: 

557 proxy_uri = proxy["uri"] 

558 if proxy_uri in deleted_entities: 

559 continue 

560 

561 delete_logic(editor, proxy_uri, graph_uri=graph_uri) 

562 deleted_entities.add(proxy_uri) 

563 

564 # Fase 2: Processa tutte le altre modifiche 

565 for change in changes: 

566 if change["action"] == "delete": 

567 subject_uri = change["subject"] 

568 predicate = change.get("predicate") 

569 object_value = change.get("object") 

570 

571 # Se stiamo eliminando un'intera entità 

572 if not predicate: 

573 if subject_uri in deleted_entities: 

574 continue 

575 

576 delete_logic(editor, subject_uri, graph_uri=graph_uri, entity_type=change.get("entity_type")) 

577 deleted_entities.add(subject_uri) 

578 # Se stiamo eliminando una tripla specifica 

579 elif object_value: 

580 # Controlla se l'oggetto è un'entità che è già stata eliminata 

581 if object_value in deleted_entities: 

582 continue 

583 

584 delete_logic(editor, subject_uri, predicate, object_value, graph_uri, change.get("entity_type")) 

585 

586 # La gestione degli orfani e dei proxy è stata spostata all'inizio del ciclo 

587 

588 elif change["action"] == "update": 

589 update_logic( 

590 editor, 

591 change["subject"], 

592 change["predicate"], 

593 change["object"], 

594 change["newObject"], 

595 graph_uri, 

596 change.get("entity_type"), 

597 ) 

598 elif change["action"] == "order": 

599 order_logic( 

600 editor, 

601 change["subject"], 

602 change["predicate"], 

603 change["object"], 

604 change["newObject"], 

605 graph_uri, 

606 temp_id_to_uri, 

607 ) 

608 

609 try: 

610 editor.save() 

611 except ValueError as ve: 

612 # Re-raise ValueError so it can be caught by the outer try-except block 

613 current_app.logger.error(f"Error during save operation: {str(ve)}") 

614 raise 

615 except Exception as save_error: 

616 current_app.logger.error(f"Error during save operation: {str(save_error)}") 

617 return jsonify( 

618 { 

619 "status": "error", 

620 "error_type": "database", 

621 "message": gettext("Failed to save changes to the database: {}").format(str(save_error)), 

622 } 

623 ), 500 

624 

625 return ( 

626 jsonify( 

627 { 

628 "status": "success", 

629 "message": gettext("Changes applied successfully"), 

630 } 

631 ), 

632 200, 

633 ) 

634 

635 except ValueError as e: 

636 # Handle validation errors specifically 

637 error_message = str(e) 

638 current_app.logger.warning(f"Validation error: {error_message}") 

639 return ( 

640 jsonify( 

641 { 

642 "status": "error", 

643 "error_type": "validation", 

644 "message": error_message, 

645 } 

646 ), 

647 400, 

648 ) 

649 except Exception as e: 

650 # Handle other errors 

651 error_message = ( 

652 f"Error while applying changes: {str(e)}\n{traceback.format_exc()}" 

653 ) 

654 current_app.logger.error(error_message) 

655 return ( 

656 jsonify( 

657 { 

658 "status": "error", 

659 "error_type": "system", 

660 "message": gettext("An error occurred while applying changes"), 

661 } 

662 ), 

663 500, 

664 ) 

665 

666 

667def get_graph_uri_from_context(graph_context): 

668 """Extract the graph URI from a graph context. 

669  

670 Args: 

671 graph_context: Either a Graph object or a direct URI reference 

672  

673 Returns: 

674 The graph URI 

675 """ 

676 if isinstance(graph_context, Graph): 

677 return graph_context.identifier 

678 else: 

679 return graph_context 

680 

681 

682def determine_datatype(value, datatype_uris): 

683 for datatype_uri in datatype_uris: 

684 validation_func = next( 

685 (d[1] for d in DATATYPE_MAPPING if str(d[0]) == str(datatype_uri)), None 

686 ) 

687 if validation_func and validation_func(value): 

688 return URIRef(datatype_uri) 

689 # If none match, default to XSD.string 

690 return XSD.string 

691 

692 

693def create_logic( 

694 editor: Editor, 

695 data: Dict[str, dict], 

696 subject=None, 

697 graph_uri=None, 

698 parent_subject=None, 

699 parent_predicate=None, 

700 temp_id_to_uri=None, 

701 parent_entity_type=None, 

702): 

703 entity_type = data.get("entity_type") 

704 properties = data.get("properties", {}) 

705 temp_id = data.get("tempId") 

706 

707 if subject is None: 

708 subject = generate_unique_uri(entity_type) 

709 

710 if temp_id and temp_id_to_uri is not None: 

711 temp_id_to_uri[temp_id] = str(subject) 

712 

713 # Create the entity type using validate_new_triple 

714 if parent_subject is not None: 

715 type_value, _, error_message = validate_new_triple( 

716 subject, RDF.type, entity_type, "create", entity_types=entity_type 

717 ) 

718 if error_message: 

719 raise ValueError(error_message) 

720 

721 if type_value is not None: 

722 editor.create(URIRef(subject), RDF.type, type_value, graph_uri) 

723 

724 # Create the relationship to the parent using validate_new_triple 

725 if parent_subject and parent_predicate: 

726 # When creating a relationship, we need to validate that the parent can have this relationship 

727 # with an entity of our type. Pass our entity_type as the object_entity_type for validation 

728 parent_value, _, error_message = validate_new_triple( 

729 parent_subject, 

730 parent_predicate, 

731 subject, 

732 "create", 

733 entity_types=parent_entity_type, 

734 ) 

735 if error_message: 

736 raise ValueError(error_message) 

737 

738 if parent_value is not None: 

739 editor.create( 

740 URIRef(parent_subject), 

741 URIRef(parent_predicate), 

742 parent_value, 

743 graph_uri, 

744 ) 

745 

746 for predicate, values in properties.items(): 

747 if not isinstance(values, list): 

748 values = [values] 

749 for value in values: 

750 if isinstance(value, dict) and "entity_type" in value: 

751 # For nested entities, create them first 

752 nested_subject = generate_unique_uri(value["entity_type"]) 

753 create_logic( 

754 editor, 

755 value, 

756 nested_subject, 

757 graph_uri, 

758 subject, 

759 predicate, 

760 temp_id_to_uri, 

761 parent_entity_type=entity_type, # Pass the current entity type as parent_entity_type 

762 ) 

763 else: 

764 # Use validate_new_triple to validate and get the correctly typed value 

765 object_value, _, error_message = validate_new_triple( 

766 subject, predicate, value, "create", entity_types=entity_type 

767 ) 

768 if error_message: 

769 raise ValueError(error_message) 

770 

771 if object_value is not None: 

772 editor.create( 

773 URIRef(subject), URIRef(predicate), object_value, graph_uri 

774 ) 

775 

776 return subject 

777 

778 

779def update_logic( 

780 editor: Editor, 

781 subject, 

782 predicate, 

783 old_value, 

784 new_value, 

785 graph_uri=None, 

786 entity_type=None, 

787): 

788 new_value, old_value, error_message = validate_new_triple( 

789 subject, predicate, new_value, "update", old_value, entity_types=entity_type 

790 ) 

791 if error_message: 

792 raise ValueError(error_message) 

793 

794 editor.update(URIRef(subject), URIRef(predicate), old_value, new_value, graph_uri) 

795 

796 

797def rebuild_entity_order( 

798 editor: Editor, 

799 ordered_by_uri: URIRef, 

800 entities: list, 

801 graph_uri=None 

802): 

803 """ 

804 Rebuild the ordering chain for a list of entities. 

805  

806 Args: 

807 editor: The editor instance 

808 ordered_by_uri: The property used for ordering 

809 entities: List of entities to be ordered 

810 graph_uri: Optional graph URI 

811 """ 

812 # First, remove all existing ordering relationships 

813 for entity in entities: 

814 for s, p, o in list(editor.g_set.triples((entity, ordered_by_uri, None))): 

815 editor.delete(entity, ordered_by_uri, o, graph_uri) 

816 

817 # Then rebuild the chain with the entities 

818 for i in range(len(entities) - 1): 

819 current_entity = entities[i] 

820 next_entity = entities[i + 1] 

821 editor.create(current_entity, ordered_by_uri, next_entity, graph_uri) 

822 

823 return editor 

824 

825 

826def delete_logic( 

827 editor: Editor, 

828 subject, 

829 predicate=None, 

830 object_value=None, 

831 graph_uri=None, 

832 entity_type=None, 

833): 

834 # Ensure we have the correct data types for all values 

835 subject_uri = URIRef(subject) 

836 predicate_uri = URIRef(predicate) if predicate else None 

837 

838 # Validate and get correctly typed object value if we have a predicate 

839 if predicate and object_value: 

840 # Use validate_new_triple to validate the deletion and get the correctly typed object 

841 _, object_value, error_message = validate_new_triple( 

842 subject, predicate, None, "delete", object_value, entity_types=entity_type 

843 ) 

844 if error_message: 

845 raise ValueError(error_message) 

846 

847 editor.delete(subject_uri, predicate_uri, object_value, graph_uri) 

848 

849 

850def order_logic( 

851 editor: Editor, 

852 subject, 

853 predicate, 

854 new_order, 

855 ordered_by, 

856 graph_uri=None, 

857 temp_id_to_uri: Optional[Dict] = None, 

858): 

859 subject_uri = URIRef(subject) 

860 predicate_uri = URIRef(predicate) 

861 ordered_by_uri = URIRef(ordered_by) 

862 # Ottieni tutte le entità ordinate attuali direttamente dall'editor 

863 current_entities = [ 

864 o for _, _, o in editor.g_set.triples((subject_uri, predicate_uri, None)) 

865 ] 

866 

867 # Dizionario per mappare le vecchie entità alle nuove 

868 old_to_new_mapping = {} 

869 

870 # Per ogni entità attuale 

871 for old_entity in current_entities: 

872 if str(old_entity) in new_order: # Processa solo le entità preesistenti 

873 # Memorizza tutte le proprietà dell'entità attuale 

874 entity_properties = list(editor.g_set.triples((old_entity, None, None))) 

875 

876 entity_type = next( 

877 (o for _, p, o in entity_properties if p == RDF.type), None 

878 ) 

879 

880 if entity_type is None: 

881 raise ValueError( 

882 f"Impossibile determinare il tipo dell'entità per {old_entity}" 

883 ) 

884 

885 # Crea una nuova entità 

886 new_entity_uri = generate_unique_uri(entity_type) 

887 old_to_new_mapping[old_entity] = new_entity_uri 

888 

889 # Cancella la vecchia entità 

890 editor.delete(subject_uri, predicate_uri, old_entity, graph_uri) 

891 editor.delete(old_entity, graph=graph_uri) 

892 

893 # Ricrea il collegamento tra il soggetto principale e la nuova entità 

894 editor.create(subject_uri, predicate_uri, new_entity_uri, graph_uri) 

895 

896 # Ripristina tutte le altre proprietà per la nuova entità 

897 for _, p, o in entity_properties: