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

1# heritrace/routes/api.py 

2 

3import traceback 

4from typing import Dict, Optional 

5 

6import validators 

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

8from flask_babel import gettext 

9from flask_login import current_user, login_required 

10from heritrace.apis.orcid import get_responsible_agent_uri 

11from heritrace.editor import Editor 

12from heritrace.extensions import (get_custom_filter, get_dataset_endpoint, 

13 get_provenance_endpoint) 

14from heritrace.services.resource_lock_manager import LockStatus 

15from heritrace.utils.datatypes import DATATYPE_MAPPING 

16from heritrace.utils.primary_source_utils import \ 

17 save_user_default_primary_source 

18from heritrace.utils.shacl_utils import determine_shape_for_classes 

19from heritrace.utils.shacl_validation import validate_new_triple 

20from heritrace.utils.sparql_utils import (find_orphaned_entities, 

21 get_available_classes, 

22 get_catalog_data, 

23 get_deleted_entities_with_filtering, 

24 import_entity_graph, 

25 import_referenced_entities) 

26from heritrace.utils.strategies import (OrphanHandlingStrategy, 

27 ProxyHandlingStrategy) 

28from heritrace.utils.uri_utils import generate_unique_uri 

29from rdflib import RDF, XSD, Graph, Literal, URIRef 

30 

31api_bp = Blueprint("api", __name__) 

32 

33 

34@api_bp.route("/catalogue") 

35@login_required 

36def catalogue_api(): 

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

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

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

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

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

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

43 

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

45 if per_page not in allowed_per_page: 

46 per_page = 50 

47 

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

49 sort_property = None 

50 

51 available_classes = get_available_classes() 

52 

53 catalog_data = get_catalog_data( 

54 selected_class=selected_class, 

55 page=page, 

56 per_page=per_page, 

57 sort_property=sort_property, 

58 sort_direction=sort_direction, 

59 selected_shape=selected_shape 

60 ) 

61 

62 catalog_data["available_classes"] = available_classes 

63 return jsonify(catalog_data) 

64 

65 

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

67@login_required 

68def get_deleted_entities_api(): 

69 """ 

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

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

72 """ 

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

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

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

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

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

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

79 

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

81 if per_page not in allowed_per_page: 

82 per_page = 50 

83 

84 deleted_entities, available_classes, selected_class, selected_shape, sortable_properties, total_count = ( 

85 get_deleted_entities_with_filtering( 

86 page, per_page, sort_property, sort_direction, selected_class, selected_shape 

87 ) 

88 ) 

89 

90 return jsonify( 

91 { 

92 "entities": deleted_entities, 

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

94 "current_page": page, 

95 "per_page": per_page, 

96 "total_count": total_count, 

97 "sort_property": sort_property, 

98 "sort_direction": sort_direction, 

99 "selected_class": selected_class, 

100 "selected_shape": selected_shape, 

101 "available_classes": available_classes, 

102 "sortable_properties": sortable_properties, 

103 } 

104 ) 

105 

106 

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

108@login_required 

109def check_lock(): 

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

111 try: 

112 data = request.get_json() 

113 resource_uri = data.get("resource_uri") 

114 

115 if not resource_uri: 

116 return ( 

117 jsonify( 

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

119 ), 

120 400, 

121 ) 

122 

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

124 

125 if status == LockStatus.LOCKED: 

126 return jsonify( 

127 { 

128 "status": "locked", 

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

130 "message": gettext( 

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

132 user=lock_info.user_name, 

133 orcid=lock_info.user_id, 

134 ), 

135 } 

136 ) 

137 elif status == LockStatus.ERROR: 

138 return ( 

139 jsonify( 

140 { 

141 "status": "error", 

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

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

144 } 

145 ), 

146 500, 

147 ) 

148 else: 

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

150 

151 except Exception as e: 

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

153 return ( 

154 jsonify( 

155 { 

156 "status": "error", 

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

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

159 } 

160 ), 

161 500, 

162 ) 

163 

164 

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

166@login_required 

167def acquire_lock(): 

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

169 try: 

170 data = request.get_json() 

171 resource_uri = data.get("resource_uri") 

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

173 

174 if not resource_uri: 

175 return ( 

176 jsonify( 

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

178 ), 

179 400, 

180 ) 

181 

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

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

184 if status == LockStatus.LOCKED: 

185 return ( 

186 jsonify( 

187 { 

188 "status": "locked", 

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

190 "message": gettext( 

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

192 user=lock_info.user_name, 

193 orcid=lock_info.user_id, 

194 ), 

195 } 

196 ), 

197 200, 

198 ) 

199 

200 # Use the provided linked_resources 

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

202 

203 if success: 

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

205 

206 return ( 

207 jsonify( 

208 { 

209 "status": "error", 

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

211 } 

212 ), 

213 423, 

214 ) 

215 

216 except Exception as e: 

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

218 return ( 

219 jsonify( 

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

221 ), 

222 500, 

223 ) 

224 

225 

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

227@login_required 

228def release_lock(): 

229 """Release a lock on a resource.""" 

230 try: 

231 data = request.get_json() 

232 resource_uri = data.get("resource_uri") 

233 

234 if not resource_uri: 

235 return ( 

236 jsonify( 

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

238 ), 

239 400, 

240 ) 

241 

242 success = g.resource_lock_manager.release_lock(resource_uri) 

243 

244 if success: 

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

246 

247 return ( 

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

249 400, 

250 ) 

251 

252 except Exception as e: 

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

254 return ( 

255 jsonify( 

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

257 ), 

258 500, 

259 ) 

260 

261 

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

263@login_required 

264def renew_lock(): 

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

266 try: 

267 data = request.get_json() 

268 resource_uri = data.get("resource_uri") 

269 

270 if not resource_uri: 

271 return ( 

272 jsonify( 

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

274 ), 

275 400, 

276 ) 

277 

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

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

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

281 

282 if success: 

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

284 

285 return ( 

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

287 423, 

288 ) 

289 

290 except Exception as e: 

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

292 return ( 

293 jsonify( 

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

295 ), 

296 500, 

297 ) 

298 

299 

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

301@login_required 

302def validate_literal(): 

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

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

305 if not value: 

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

307 

308 matching_datatypes = [] 

309 for datatype, validation_func, _ in DATATYPE_MAPPING: 

310 if validation_func(value): 

311 matching_datatypes.append(str(datatype)) 

312 

313 if not matching_datatypes: 

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

315 

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

317 

318 

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

320@login_required 

321def check_orphans(): 

322 """ 

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

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

325 """ 

326 try: 

327 # Get strategies from configuration 

328 orphan_strategy = current_app.config.get( 

329 "ORPHAN_HANDLING_STRATEGY", OrphanHandlingStrategy.KEEP 

330 ) 

331 proxy_strategy = current_app.config.get( 

332 "PROXY_HANDLING_STRATEGY", ProxyHandlingStrategy.KEEP 

333 ) 

334 

335 data = request.json 

336 # Validate required fields 

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

338 return ( 

339 jsonify( 

340 { 

341 "status": "error", 

342 "error_type": "validation", 

343 "message": gettext( 

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

345 ), 

346 } 

347 ), 

348 400, 

349 ) 

350 

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

352 entity_type = data.get("entity_type") 

353 entity_shape = data.get("entity_shape") 

354 custom_filter = get_custom_filter() 

355 

356 orphans = [] 

357 intermediate_orphans = [] 

358 

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

360 check_for_orphans = orphan_strategy in ( 

361 OrphanHandlingStrategy.DELETE, 

362 OrphanHandlingStrategy.ASK, 

363 ) 

364 check_for_proxies = proxy_strategy in ( 

365 ProxyHandlingStrategy.DELETE, 

366 ProxyHandlingStrategy.ASK, 

367 ) 

368 if check_for_orphans or check_for_proxies: 

369 for change in changes: 

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

371 found_orphans, found_intermediates = find_orphaned_entities( 

372 change["subject"], 

373 entity_type, 

374 change.get("predicate"), 

375 change.get("object"), 

376 ) 

377 # Only collect orphans if we need to handle them 

378 if check_for_orphans: 

379 orphans.extend(found_orphans) 

380 

381 # Only collect proxies if we need to handle them 

382 if check_for_proxies: 

383 intermediate_orphans.extend(found_intermediates) 

384 

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

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

387 proxy_strategy == ProxyHandlingStrategy.KEEP or not intermediate_orphans 

388 ): 

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

390 

391 # Format entities for display 

392 def format_entities(entities, is_intermediate=False): 

393 return [ 

394 { 

395 "uri": entity["uri"], 

396 "label": custom_filter.human_readable_entity( 

397 entity["uri"], (entity["type"], entity_shape) 

398 ), 

399 "type": custom_filter.human_readable_class( 

400 (entity["type"], entity_shape) 

401 ), 

402 "is_intermediate": is_intermediate, 

403 } 

404 for entity in entities 

405 ] 

406 

407 # Create a unified list of affected entities 

408 affected_entities = format_entities(orphans) + format_entities( 

409 intermediate_orphans, is_intermediate=True 

410 ) 

411 

412 # Determine if we should automatically delete entities 

413 should_delete_orphans = orphan_strategy == OrphanHandlingStrategy.DELETE 

414 should_delete_proxies = proxy_strategy == ProxyHandlingStrategy.DELETE 

415 

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

417 if should_delete_orphans and should_delete_proxies: 

418 return jsonify( 

419 { 

420 "status": "success", 

421 "affected_entities": affected_entities, 

422 "should_delete": True, 

423 "orphan_strategy": orphan_strategy.value, 

424 "proxy_strategy": proxy_strategy.value, 

425 } 

426 ) 

427 

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

429 return jsonify( 

430 { 

431 "status": "success", 

432 "affected_entities": affected_entities, 

433 "should_delete": False, 

434 "orphan_strategy": orphan_strategy.value, 

435 "proxy_strategy": proxy_strategy.value, 

436 } 

437 ) 

438 except ValueError as e: 

439 # Handle validation errors specifically 

440 error_message = str(e) 

441 current_app.logger.warning( 

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

443 ) 

444 return ( 

445 jsonify( 

446 { 

447 "status": "error", 

448 "error_type": "validation", 

449 "message": gettext( 

450 "An error occurred while checking for orphaned entities" 

451 ), 

452 } 

453 ), 

454 400, 

455 ) 

456 except Exception as e: 

457 # Handle other errors 

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

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

460 return ( 

461 jsonify( 

462 { 

463 "status": "error", 

464 "error_type": "system", 

465 "message": gettext( 

466 "An error occurred while checking for orphaned entities" 

467 ), 

468 } 

469 ), 

470 500, 

471 ) 

472 

473 

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

475@login_required 

476def apply_changes(): 

477 """Apply changes to entities. 

478 

479 Request body: 

480 { 

481 "subject": (str) Main entity URI being modified, 

482 "changes": (list) List of changes to apply, 

483 "primary_source": (str) Primary source to use for provenance, 

484 "save_default_source": (bool) Whether to save primary_source as default for current user, 

485 "affected_entities": (list) Entities potentially affected by delete operations, 

486 "delete_affected": (bool) Whether to delete affected entities 

487 } 

488 

489 Responses: 

490 200 OK: Changes applied successfully 

491 400 Bad Request: Invalid request or validation error 

492 500 Internal Server Error: Server error while applying changes 

493 """ 

494 try: 

495 changes = request.get_json() 

496 

497 if not changes: 

498 return jsonify({"error": "No request data provided"}), 400 

499 

500 first_change = changes[0] if changes else {} 

501 subject = first_change.get("subject") 

502 affected_entities = first_change.get("affected_entities", []) 

503 delete_affected = first_change.get("delete_affected", False) 

504 primary_source = first_change.get("primary_source") 

505 save_default_source = first_change.get("save_default_source", False) 

506 

507 if primary_source and not validators.url(primary_source): 

508 return jsonify({"error": "Invalid primary source URL"}), 400 

509 

510 if save_default_source and primary_source and validators.url(primary_source): 

511 save_user_default_primary_source(current_user.orcid, primary_source) 

512 

513 deleted_entities = set() 

514 editor = Editor( 

515 get_dataset_endpoint(), 

516 get_provenance_endpoint(), 

517 current_app.config["COUNTER_HANDLER"], 

518 URIRef(get_responsible_agent_uri(current_user.orcid)), 

519 current_app.config["PRIMARY_SOURCE"], 

520 current_app.config["DATASET_GENERATION_TIME"], 

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

522 ) 

523 

524 if primary_source and validators.url(primary_source): 

525 editor.set_primary_source(primary_source) 

526 

527 has_entity_deletion = any( 

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

529 for change in changes 

530 ) 

531 

532 editor = import_entity_graph( 

533 editor, 

534 subject, 

535 include_referencing_entities=has_entity_deletion 

536 ) 

537 

538 for change in changes: 

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

540 data = change.get("data") 

541 if data: 

542 import_referenced_entities(editor, data) 

543 

544 editor.preexisting_finished() 

545 

546 graph_uri = None 

547 if editor.dataset_is_quadstore: 

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

549 graph_context = quad[3] 

550 graph_uri = get_graph_uri_from_context(graph_context) 

551 break 

552 

553 temp_id_to_uri = {} 

554 for change in changes: 

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

556 data = change.get("data") 

557 if data: 

558 subject = create_logic( 

559 editor, 

560 data, 

561 subject, 

562 graph_uri, 

563 temp_id_to_uri=temp_id_to_uri, 

564 parent_entity_type=None, 

565 ) 

566 

567 orphan_strategy = current_app.config.get( 

568 "ORPHAN_HANDLING_STRATEGY", OrphanHandlingStrategy.KEEP 

569 ) 

570 proxy_strategy = current_app.config.get( 

571 "PROXY_HANDLING_STRATEGY", ProxyHandlingStrategy.KEEP 

572 ) 

573 # Separiamo le operazioni di delete in due fasi: 

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

575 # 2. Poi eliminiamo le triple specifiche 

576 

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

578 if affected_entities and delete_affected: 

579 # Separa gli orfani dalle entità proxy 

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

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

582 

583 # Gestione degli orfani secondo la strategia per gli orfani 

584 should_delete_orphans = ( 

585 orphan_strategy == OrphanHandlingStrategy.DELETE 

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

587 ) 

588 

589 if should_delete_orphans and orphans: 

590 for orphan in orphans: 

591 orphan_uri = orphan["uri"] 

592 if orphan_uri in deleted_entities: 

593 continue 

594 

595 delete_logic(editor, orphan_uri, graph_uri=graph_uri) 

596 deleted_entities.add(orphan_uri) 

597 

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

599 should_delete_proxies = ( 

600 proxy_strategy == ProxyHandlingStrategy.DELETE 

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

602 ) 

603 

604 if should_delete_proxies and proxies: 

605 for proxy in proxies: 

606 proxy_uri = proxy["uri"] 

607 if proxy_uri in deleted_entities: 

608 continue 

609 

610 delete_logic(editor, proxy_uri, graph_uri=graph_uri) 

611 deleted_entities.add(proxy_uri) 

612 

613 # Fase 2: Processa tutte le altre modifiche 

614 for change in changes: 

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

616 subject_uri = change["subject"] 

617 predicate = change.get("predicate") 

618 object_value = change.get("object") 

619 

620 # Se stiamo eliminando un'intera entità 

621 if not predicate: 

622 if subject_uri in deleted_entities: 

623 continue 

624 

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

626 deleted_entities.add(subject_uri) 

627 # Se stiamo eliminando una tripla specifica 

628 elif object_value: 

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

630 if object_value in deleted_entities: 

631 continue 

632 

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

634 

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

636 

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

638 update_logic( 

639 editor, 

640 change["subject"], 

641 change["predicate"], 

642 change["object"], 

643 change["newObject"], 

644 graph_uri, 

645 change.get("entity_type"), 

646 change.get("entity_shape"), 

647 ) 

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

649 order_logic( 

650 editor, 

651 change["subject"], 

652 change["predicate"], 

653 change["object"], 

654 change["newObject"], 

655 graph_uri, 

656 temp_id_to_uri, 

657 ) 

658 

659 try: 

660 editor.save() 

661 except ValueError as ve: 

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

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

664 raise 

665 except Exception as save_error: 

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

667 return jsonify( 

668 { 

669 "status": "error", 

670 "error_type": "database", 

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

672 } 

673 ), 500 

674 

675 return ( 

676 jsonify( 

677 { 

678 "status": "success", 

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

680 } 

681 ), 

682 200, 

683 ) 

684 

685 except ValueError as e: 

686 # Handle validation errors specifically 

687 error_message = str(e) 

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

689 return ( 

690 jsonify( 

691 { 

692 "status": "error", 

693 "error_type": "validation", 

694 "message": error_message, 

695 } 

696 ), 

697 400, 

698 ) 

699 except Exception as e: 

700 # Handle other errors 

701 error_message = ( 

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

703 ) 

704 current_app.logger.error(error_message) 

705 return ( 

706 jsonify( 

707 { 

708 "status": "error", 

709 "error_type": "system", 

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

711 } 

712 ), 

713 500, 

714 ) 

715 

716 

717def get_graph_uri_from_context(graph_context): 

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

719  

720 Args: 

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

722  

723 Returns: 

724 The graph URI 

725 """ 

726 if isinstance(graph_context, Graph): 

727 return graph_context.identifier 

728 else: 

729 return graph_context 

730 

731 

732def determine_datatype(value, datatype_uris): 

733 for datatype_uri in datatype_uris: 

734 validation_func = next( 

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

736 ) 

737 if validation_func and validation_func(value): 

738 return URIRef(datatype_uri) 

739 # If none match, default to XSD.string 

740 return XSD.string 

741 

742 

743def create_logic( 

744 editor: Editor, 

745 data: Dict[str, dict], 

746 subject=None, 

747 graph_uri=None, 

748 parent_subject=None, 

749 parent_predicate=None, 

750 temp_id_to_uri=None, 

751 parent_entity_type=None, 

752): 

753 """ 

754 Recursively creates an entity and its properties based on a dictionary. 

755 

756 This function handles the creation of a main entity and any nested entities 

757 defined within its properties. It validates each triple before creation and 

758 can link the new entity to a parent entity. 

759 

760 Args: 

761 editor (Editor): The editor instance for graph operations. 

762 data (Dict[str, dict]): A dictionary describing the entity to create. 

763 subject (URIRef, optional): The subject URI of the entity. If None, a new URI is generated. 

764 graph_uri (str, optional): The named graph URI for the operations. 

765 parent_subject (URIRef, optional): The subject URI of the parent entity. 

766 parent_predicate (URIRef, optional): The predicate URI linking the parent to this entity. 

767 temp_id_to_uri (Dict, optional): A dictionary mapping temporary frontend IDs to backend URIs. 

768 parent_entity_type (str, optional): The RDF type of the parent entity, for validation. 

769 

770 Example of `data` structure: 

771 { 

772 "entity_type": "http://purl.org/spar/fabio/JournalArticle", 

773 "properties": { 

774 "http://purl.org/spar/pro/isDocumentContextFor": [ 

775 { 

776 "entity_type": "http://purl.org/spar/pro/RoleInTime", 

777 "properties": { 

778 "http://purl.org/spar/pro/isHeldBy": [ 

779 "https://w3id.org/oc/meta/ra/09110374" 

780 ], 

781 "http://purl.org/spar/pro/withRole": "http://purl.org/spar/pro/author" 

782 }, 

783 "tempId": "temp-1" 

784 } 

785 ] 

786 } 

787 } 

788 """ 

789 entity_type = data.get("entity_type") 

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

791 temp_id = data.get("tempId") 

792 

793 if subject is None: 

794 subject = generate_unique_uri(entity_type) 

795 

796 if temp_id and temp_id_to_uri is not None: 

797 temp_id_to_uri[temp_id] = str(subject) 

798 

799 # Create the entity type using validate_new_triple 

800 if parent_subject is not None: 

801 type_value, _, error_message = validate_new_triple( 

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

803 ) 

804 if error_message: 

805 raise ValueError(error_message) 

806 

807 if type_value is not None: 

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

809 

810 # Create the relationship to the parent using validate_new_triple 

811 if parent_subject and parent_predicate: 

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

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

814 parent_value, _, error_message = validate_new_triple( 

815 parent_subject, 

816 parent_predicate, 

817 subject, 

818 "create", 

819 entity_types=parent_entity_type, 

820 ) 

821 if error_message: 

822 raise ValueError(error_message) 

823 

824 if parent_value is not None: 

825 editor.create( 

826 URIRef(parent_subject), 

827 URIRef(parent_predicate), 

828 parent_value, 

829 graph_uri, 

830 ) 

831 

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

833 if not isinstance(values, list): 

834 values = [values] 

835 for value in values: 

836 # CASE 1: Nested Entity. 

837 # If the value is a dictionary containing 'entity_type', it's a nested entity 

838 # that needs to be created recursively. 

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

840 # A new URI is generated for the nested entity. 

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

842 # The function calls itself to create the nested entity. The current entity 

843 # becomes the parent for the nested one. 

844 create_logic( 

845 editor, 

846 value, 

847 nested_subject, 

848 graph_uri, 

849 subject, # Current entity is the parent subject 

850 predicate, # The predicate linking parent to child 

851 temp_id_to_uri, 

852 parent_entity_type=entity_type, 

853 ) 

854 # CASE 2: Existing Entity Reference. 

855 elif isinstance(value, dict) and value.get("is_existing_entity", False): 

856 entity_uri = value.get("entity_uri") 

857 if entity_uri: 

858 object_value = URIRef(entity_uri) 

859 editor.create( 

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

861 ) 

862 else: 

863 raise ValueError("Missing entity_uri in existing entity reference") 

864 # CASE 3: Custom Property. 

865 # If the value is a dictionary marked as 'is_custom_property', it's a property 

866 # that is not defined in the SHACL shape and should be added without validation. 

867 elif isinstance(value, dict) and value.get("is_custom_property", False): 

868 # The property is created directly based on its type ('uri' or 'literal'). 

869 if value["type"] == "uri": 

870 object_value = URIRef(value["value"]) 

871 elif value["type"] == "literal": 

872 datatype = ( 

873 URIRef(value["datatype"]) 

874 if "datatype" in value 

875 else XSD.string 

876 ) 

877 object_value = Literal(value["value"], datatype=datatype) 

878 else: 

879 raise ValueError(f"Unknown custom property type: {value['type']}") 

880 

881 editor.create( 

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

883 ) 

884 # CASE 4: Standard Property. 

885 # This is the default case for all other properties. The value can be a 

886 # simple literal (e.g., a string, number) or a URI string. 

887 else: 

888 # The value is validated against the SHACL shape for the current entity type. 

889 # `validate_new_triple` checks if the triple is valid and returns the 

890 # correctly typed RDF object (e.g., Literal, URIRef). 

891 object_value, _, error_message = validate_new_triple( 

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

893 ) 

894 if error_message: 

895 raise ValueError(error_message) 

896 

897 if object_value is not None: 

898 editor.create( 

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

900 ) 

901 

902 return subject 

903