Coverage for heritrace/utils/display_rules_utils.py: 94%

1# SPDX-FileCopyrightText: 2024-2025 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import logging 

8from collections import OrderedDict 

9from dataclasses import dataclass 

10from typing import TYPE_CHECKING 

11from urllib.parse import unquote 

12 

13from pyparsing.exceptions import ParseException 

14from rdflib import Graph, Literal, URIRef 

15from rdflib.plugins.sparql.algebra import translateQuery, traverse 

16from rdflib.plugins.sparql.parser import parseQuery 

17from SPARQLWrapper import JSON 

18 

19from heritrace.extensions import ( 

20 get_custom_filter, 

21 get_display_rules, 

22 get_form_fields, 

23 get_sparql, 

24 get_sparql_bindings, 

25 select_results, 

26) 

27 

28if TYPE_CHECKING: 

29 from rdflib.query import ResultRow 

30 

31 

32@dataclass(slots=True) 

33class GroupingContext: 

34 subject: URIRef 

35 triples: list[tuple[URIRef, URIRef, URIRef | Literal]] 

36 grouped_triples: OrderedDict 

37 fetched_values_map: dict[str, str] 

38 relevant_properties: set 

39 historical_snapshot: Graph | None 

40 highest_priority_class: str | None 

41 highest_priority_shape: str | None 

42 

43 

44_SUBJECT_LABEL_PAIR_LENGTH = 2 

45 

46_INVERSE_PROBE = URIRef("urn:heritrace:inverse-probe") 

47 

48 

49def _binds_uri_as_object(fetch_uri_display: str) -> bool: 

50 probed = fetch_uri_display.replace("[[uri]]", f"<{_INVERSE_PROBE}>") 

51 try: 

52 algebra = translateQuery(parseQuery(probed)).algebra 

53 except ParseException: 

54 # rdflib evaluates fetchUriDisplay against the snapshot graph with this 

55 # same parser, so a query it cannot parse cannot produce a label there 

56 # and cannot need incoming references either. 

57 logging.getLogger(__name__).warning( 

58 "fetchUriDisplay query is not valid SPARQL and will not render in " 

59 "entity history: %s", 

60 fetch_uri_display, 

61 ) 

62 return False 

63 found = False 

64 

65 def visit(node: object) -> None: 

66 nonlocal found 

67 triples = getattr(node, "triples", None) 

68 if triples is None: 

69 return 

70 for triple in triples: 

71 if triple[2] == _INVERSE_PROBE: 

72 found = True 

73 

74 traverse(algebra, visitPre=visit) 

75 return found 

76 

77 

78# A rule that binds [[uri]] as an object resolves its label from entities 

79# pointing at the current one, so history retrieval must collect reverse 

80# relations to render it. That means reconstructing the history of every 

81# referring entity, so it is done only when a rule actually needs it. 

82def uses_inverse_relations(display_rules: list[dict]) -> bool: 

83 return any( 

84 _binds_uri_as_object(rule["fetchUriDisplay"]) 

85 for rule in display_rules 

86 if "fetchUriDisplay" in rule 

87 ) 

88 

89 

90def find_matching_rule( 

91 class_uri: str | None = None, 

92 shape_uri: str | None = None, 

93 rules: list[dict] | None = None, 

94) -> dict | None: 

95 """ 

96 Find the most appropriate rule for a given class and/or shape. 

97 At least one of class_uri or shape_uri must be provided. 

98 

99 Args: 

100 class_uri: Optional URI of the class 

101 shape_uri: Optional URI of the shape 

102 rules: Optional list of rules to search in, defaults to global display_rules 

103 

104 Returns: 

105 The matching rule or None if no match is found 

106 """ 

107 if not rules: 

108 rules = get_display_rules() 

109 if not rules: 

110 return None 

111 

112 # Initialize variables to track potential matches 

113 class_match = None 

114 shape_match = None 

115 highest_priority = float("inf") 

116 

117 # Scan all rules to find the best match based on priority 

118 for rule in rules: 

119 rule_priority = rule.get("priority", 0) 

120 

121 # Case 1: Both class and shape match (exact match) 

122 if ( 

123 class_uri 

124 and shape_uri 

125 and "class" in rule["target"] 

126 and rule["target"]["class"] == str(class_uri) 

127 and "shape" in rule["target"] 

128 and rule["target"]["shape"] == str(shape_uri) 

129 ): 

130 # Exact match always takes highest precedence 

131 return rule 

132 

133 # Case 2: Only class matches 

134 if ( 

135 class_uri 

136 and "class" in rule["target"] 

137 and rule["target"]["class"] == str(class_uri) 

138 and "shape" not in rule["target"] 

139 ): 

140 if class_match is None or rule_priority < highest_priority: 

141 class_match = rule 

142 highest_priority = rule_priority 

143 

144 # Case 3: Only shape matches 

145 elif ( 

146 shape_uri 

147 and "shape" in rule["target"] 

148 and rule["target"]["shape"] == str(shape_uri) 

149 and "class" not in rule["target"] 

150 ) and (shape_match is None or rule_priority < highest_priority): 

151 shape_match = rule 

152 highest_priority = rule_priority 

153 

154 # Return the best match based on priority 

155 # Shape rules typically have higher specificity, 

156 # so prefer them if they have equal priority 

157 if shape_match and ( 

158 class_match is None 

159 or shape_match.get("priority", 0) <= class_match.get("priority", 0) 

160 ): 

161 return shape_match 

162 if class_match: 

163 return class_match 

164 

165 return None 

166 

167 

168def get_class_priority(entity_key: tuple[str, str | None]) -> float: 

169 """ 

170 Returns the priority of a specific entity key (class_uri, shape_uri). 

171 Calculates the priority directly from the display rules. 

172 Classes without defined rules receive the lowest priority (highest number). 

173 

174 Args: 

175 entity_key: A tuple (class_uri, shape_uri) 

176 """ 

177 class_uri = entity_key[0] 

178 shape_uri = entity_key[1] 

179 

180 rule = find_matching_rule(class_uri, shape_uri) 

181 return rule.get("priority", 0) if rule else float("inf") 

182 

183 

184def is_entity_type_visible(entity_key: tuple[str, str | None]) -> bool: 

185 """ 

186 Determines if an entity type should be displayed. 

187 

188 Args: 

189 entity_key: A tuple (class_uri, shape_uri) 

190 """ 

191 class_uri = entity_key[0] 

192 shape_uri = entity_key[1] 

193 

194 rule = find_matching_rule(class_uri, shape_uri) 

195 return rule.get("shouldBeDisplayed", True) if rule else True 

196 

197 

198def get_sortable_properties(entity_key: tuple[str, str | None]) -> list[dict[str, str]]: 

199 """ 

200 Gets the sortable properties from display rules for an entity type and/or shape. 

201 Infers the sorting type from form_fields_cache. 

202 

203 Args: 

204 entity_key: A tuple (class_uri, shape_uri) 

205 

206 Returns: 

207 List of dictionaries with sorting information 

208 """ 

209 display_rules = get_display_rules() 

210 if not display_rules: 

211 return [] 

212 

213 form_fields = get_form_fields() 

214 

215 class_uri = entity_key[0] 

216 shape_uri = entity_key[1] 

217 

218 rule = find_matching_rule(class_uri, shape_uri, display_rules) 

219 if not rule or "sortableBy" not in rule: 

220 return [] 

221 

222 sort_props = [] 

223 for sort_config in rule["sortableBy"]: 

224 prop = sort_config.copy() 

225 

226 for display_prop in rule["displayProperties"]: 

227 if display_prop["property"] == prop["property"]: 

228 if "displayRules" in display_prop: 

229 prop["displayName"] = display_prop["displayRules"][0]["displayName"] 

230 else: 

231 prop["displayName"] = display_prop.get( 

232 "displayName", prop["property"] 

233 ) 

234 break 

235 

236 # Default to string sorting 

237 prop["sortType"] = "string" 

238 

239 # Try to determine the sort type from form fields 

240 if form_fields and ( 

241 entity_key in form_fields and prop["property"] in form_fields[entity_key] 

242 ): 

243 field_info = form_fields[entity_key][prop["property"]][ 

244 0 

245 ] # Take the first field definition 

246 prop["sortType"] = determine_sort_type(field_info) 

247 

248 sort_props.append(prop) 

249 

250 return sort_props 

251 

252 

253def determine_sort_type(field_info: dict) -> str: 

254 """Helper function to determine sort type from field info.""" 

255 # If there's a shape, it's a reference to an entity (sort by label) 

256 if field_info.get("nodeShape"): 

257 return "string" 

258 # Otherwise look at the datatypes 

259 if field_info.get("datatypes"): 

260 datatype = str(field_info["datatypes"][0]).lower() 

261 if any(t in datatype for t in ["date", "time"]): 

262 return "date" 

263 if any(t in datatype for t in ["int", "float", "decimal", "double", "number"]): 

264 return "number" 

265 if "boolean" in datatype: 

266 return "boolean" 

267 # Default to string 

268 return "string" 

269 

270 

271def get_highest_priority_class(subject_classes: list[str]) -> str | None: 

272 """ 

273 Find the highest priority class from the given list of classes. 

274 

275 Args: 

276 subject_classes: List of class URIs 

277 

278 Returns: 

279 The highest priority class or None if no classes are provided 

280 """ 

281 from heritrace.utils.shacl_utils import determine_shape_for_classes # noqa: PLC0415 

282 

283 if not subject_classes: 

284 return None 

285 

286 highest_priority = float("inf") 

287 highest_priority_class = None 

288 

289 for raw_class_uri in subject_classes: 

290 class_uri = str(raw_class_uri) 

291 shape = determine_shape_for_classes([class_uri]) 

292 entity_key = (class_uri, shape) 

293 priority = get_class_priority(entity_key) 

294 if priority < highest_priority: 

295 highest_priority = priority 

296 highest_priority_class = class_uri 

297 

298 if highest_priority_class is None and subject_classes: 

299 highest_priority_class = str(subject_classes[0]) 

300 

301 return highest_priority_class 

302 

303 

304def _ensure_grouped_entry( 

305 ctx: GroupingContext, 

306 display_name: str, 

307 prop_uri: str, 

308 object_shape: str | None, 

309) -> None: 

310 if display_name not in ctx.grouped_triples: 

311 ctx.grouped_triples[display_name] = { 

312 "property": prop_uri, 

313 "triples": [], 

314 "subjectClass": ctx.highest_priority_class, 

315 "subjectShape": ctx.highest_priority_shape, 

316 "objectShape": object_shape, 

317 } 

318 

319 

320def _apply_ordering_to_group( 

321 ctx: GroupingContext, 

322 current_prop_config: dict, 

323 order_property: str | None, 

324 display_name: str, 

325) -> None: 

326 ctx.grouped_triples[display_name]["is_draggable"] = True 

327 ctx.grouped_triples[display_name]["ordered_by"] = order_property 

328 process_ordering( 

329 ctx, 

330 current_prop_config, 

331 order_property, 

332 display_name, 

333 ) 

334 

335 

336def _apply_intermediate_relation( 

337 grouped_triples: OrderedDict, 

338 display_name: str, 

339 *sources: dict, 

340) -> None: 

341 for source in sources: 

342 if "intermediateRelation" in source: 

343 grouped_triples[display_name]["intermediateRelation"] = source[ 

344 "intermediateRelation" 

345 ] 

346 return 

347 

348 

349def _process_property_with_nested_display_rules( 

350 prop_uri: str, 

351 current_prop_config: dict, 

352 ctx: GroupingContext, 

353) -> None: 

354 is_ordered = "orderedBy" in current_prop_config 

355 order_property = current_prop_config.get("orderedBy") 

356 

357 for display_rule_nested in current_prop_config["displayRules"]: 

358 display_name_nested = display_rule_nested.get("displayName", prop_uri) 

359 ctx.relevant_properties.add(prop_uri) 

360 object_shape = display_rule_nested.get("shape") 

361 if current_prop_config.get("isVirtual"): 

362 process_virtual_property_display( 

363 display_name_nested, 

364 current_prop_config, 

365 ctx, 

366 ) 

367 else: 

368 process_display_rule( 

369 display_name_nested, 

370 prop_uri, 

371 display_rule_nested, 

372 ctx, 

373 object_shape=object_shape, 

374 ) 

375 if is_ordered and not current_prop_config.get("isVirtual", False): 

376 _apply_ordering_to_group( 

377 ctx, 

378 current_prop_config, 

379 order_property, 

380 display_name_nested, 

381 ) 

382 

383 _ensure_grouped_entry( 

384 ctx, 

385 display_name_nested, 

386 prop_uri, 

387 display_rule_nested.get("shape"), 

388 ) 

389 

390 _apply_intermediate_relation( 

391 ctx.grouped_triples, 

392 display_name_nested, 

393 display_rule_nested, 

394 current_prop_config, 

395 ) 

396 

397 

398def _process_property_with_simple_config( 

399 prop_uri: str, 

400 current_prop_config: dict, 

401 current_form_field: list[dict] | None, 

402 ctx: GroupingContext, 

403) -> None: 

404 display_name_simple = current_prop_config.get("displayName", prop_uri) 

405 # Only add non-virtual properties to relevant_properties 

406 # Virtual properties are handled separately in entity.py 

407 if not current_prop_config.get("isVirtual"): 

408 ctx.relevant_properties.add(prop_uri) 

409 

410 object_shape = None 

411 if current_form_field: 

412 for form_field in current_form_field: 

413 object_shape = form_field.get("nodeShape") 

414 break 

415 

416 if current_prop_config.get("isVirtual"): 

417 process_virtual_property_display( 

418 display_name_simple, 

419 current_prop_config, 

420 ctx, 

421 ) 

422 else: 

423 process_display_rule( 

424 display_name_simple, 

425 prop_uri, 

426 current_prop_config, 

427 ctx, 

428 object_shape=object_shape, 

429 ) 

430 if "orderedBy" in current_prop_config and not current_prop_config.get( 

431 "isVirtual", False 

432 ): 

433 _ensure_grouped_entry( 

434 ctx, 

435 display_name_simple, 

436 prop_uri, 

437 current_prop_config.get("shape"), 

438 ) 

439 _apply_ordering_to_group( 

440 ctx, 

441 current_prop_config, 

442 current_prop_config.get("orderedBy"), 

443 display_name_simple, 

444 ) 

445 if "intermediateRelation" in current_prop_config: 

446 _ensure_grouped_entry( 

447 ctx, 

448 display_name_simple, 

449 prop_uri, 

450 current_prop_config.get("shape"), 

451 ) 

452 ctx.grouped_triples[display_name_simple]["intermediateRelation"] = ( 

453 current_prop_config["intermediateRelation"] 

454 ) 

455 

456 

457def _process_property_with_display_rules( 

458 prop_uri: str, 

459 matching_rule: dict, 

460 matching_form_field: dict | None, 

461 ctx: GroupingContext, 

462) -> None: 

463 current_prop_config = None 

464 for prop_config in matching_rule.get("displayProperties", []): 

465 config_identifier = ( 

466 prop_config.get("displayName") 

467 if prop_config.get("isVirtual") 

468 else prop_config.get("property") 

469 ) 

470 if config_identifier == prop_uri: 

471 current_prop_config = prop_config 

472 break 

473 

474 current_form_field = ( 

475 matching_form_field.get(prop_uri) if matching_form_field else None 

476 ) 

477 

478 if current_prop_config: 

479 if "displayRules" in current_prop_config: 

480 _process_property_with_nested_display_rules( 

481 prop_uri, 

482 current_prop_config, 

483 ctx, 

484 ) 

485 else: 

486 _process_property_with_simple_config( 

487 prop_uri, 

488 current_prop_config, 

489 current_form_field, 

490 ctx, 

491 ) 

492 else: 

493 # Property without specific configuration - add to relevant_properties 

494 # Don't process properties without config 

495 # (they are not virtual in this case) 

496 ctx.relevant_properties.add(prop_uri) 

497 process_default_property( 

498 prop_uri, 

499 ctx.triples, 

500 ctx.grouped_triples, 

501 ctx.highest_priority_shape, 

502 ctx.highest_priority_class, 

503 ) 

504 

505 

506def get_grouped_triples( 

507 subject: URIRef, 

508 triples: list[tuple[URIRef, URIRef, URIRef | Literal]], 

509 valid_predicates_info: list[str], 

510 historical_snapshot: Graph | None = None, 

511 entity_key: tuple[str | None, str | None] = (None, None), 

512) -> tuple[OrderedDict, set]: 

513 highest_priority_class, highest_priority_shape = entity_key 

514 display_rules = get_display_rules() 

515 form_fields = get_form_fields() 

516 

517 grouped_triples = OrderedDict() 

518 relevant_properties: set = set() 

519 fetched_values_map: dict[str, str] = {} 

520 

521 ctx = GroupingContext( 

522 subject=subject, 

523 triples=triples, 

524 grouped_triples=grouped_triples, 

525 fetched_values_map=fetched_values_map, 

526 relevant_properties=relevant_properties, 

527 historical_snapshot=historical_snapshot, 

528 highest_priority_class=highest_priority_class, 

529 highest_priority_shape=highest_priority_shape, 

530 ) 

531 

532 matching_rule = find_matching_rule( 

533 highest_priority_class, highest_priority_shape, display_rules 

534 ) 

535 matching_form_field = form_fields.get( 

536 (highest_priority_class, highest_priority_shape) 

537 ) 

538 

539 ordered_properties = [] 

540 if display_rules and matching_rule: 

541 for prop_config in matching_rule.get("displayProperties", []): 

542 if prop_config.get("isVirtual"): 

543 prop_uri = prop_config.get("displayName") 

544 else: 

545 prop_uri = prop_config.get("property") 

546 if prop_uri and prop_uri not in ordered_properties: 

547 ordered_properties.append(prop_uri) 

548 

549 for prop_uri in valid_predicates_info: 

550 if prop_uri not in ordered_properties: 

551 ordered_properties.append(prop_uri) 

552 

553 for prop_uri in ordered_properties: 

554 if display_rules and matching_rule: 

555 _process_property_with_display_rules( 

556 prop_uri, 

557 matching_rule, 

558 matching_form_field, 

559 ctx, 

560 ) 

561 else: 

562 # No display rules or no matching rule - 

563 # add all properties to relevant_properties 

564 ctx.relevant_properties.add(prop_uri) 

565 process_default_property( 

566 prop_uri, 

567 ctx.triples, 

568 ctx.grouped_triples, 

569 ctx.highest_priority_shape, 

570 ctx.highest_priority_class, 

571 ) 

572 

573 ctx.grouped_triples = OrderedDict(ctx.grouped_triples) 

574 return ctx.grouped_triples, ctx.relevant_properties 

575 

576 

577def process_display_rule( 

578 display_name: str, 

579 prop_uri: str, 

580 rule: dict, 

581 ctx: GroupingContext, 

582 object_shape: str | None = None, 

583) -> None: 

584 if display_name not in ctx.grouped_triples: 

585 ctx.grouped_triples[display_name] = { 

586 "property": prop_uri, 

587 "triples": [], 

588 "subjectClass": ctx.highest_priority_class, 

589 "subjectShape": ctx.highest_priority_shape, 

590 "objectShape": object_shape, 

591 "intermediateRelation": rule.get("intermediateRelation"), 

592 } 

593 for triple in ctx.triples: 

594 if str(triple[1]) == prop_uri: 

595 if rule.get("fetchValueFromQuery"): 

596 if ctx.historical_snapshot: 

597 result, external_entity = execute_historical_query( 

598 rule["fetchValueFromQuery"], 

599 ctx.subject, 

600 triple[2], 

601 ctx.historical_snapshot, 

602 ) 

603 else: 

604 result, external_entity = execute_sparql_query( 

605 rule["fetchValueFromQuery"], ctx.subject, triple[2] 

606 ) 

607 if result: 

608 ctx.fetched_values_map[str(result)] = str(triple[2]) 

609 new_triple = (str(triple[0]), str(triple[1]), str(result)) 

610 object_uri = str(triple[2]) 

611 new_triple_data = { 

612 "triple": new_triple, 

613 "external_entity": external_entity, 

614 "object": object_uri, 

615 "subjectClass": ctx.highest_priority_class, 

616 "subjectShape": ctx.highest_priority_shape, 

617 "objectShape": object_shape, 

618 } 

619 ctx.grouped_triples[display_name]["triples"].append(new_triple_data) 

620 else: 

621 if str(triple[1]) == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type": 

622 from heritrace.utils.shacl_utils import ( # noqa: PLC0415 

623 determine_shape_for_classes, 

624 ) 

625 

626 object_class_shape = determine_shape_for_classes([triple[2]]) 

627 result = get_custom_filter().human_readable_class( 

628 (triple[2], object_class_shape) 

629 ) 

630 else: 

631 result = triple[2] 

632 

633 object_uri = str(triple[2]) 

634 

635 new_triple_data = { 

636 "triple": (str(triple[0]), str(triple[1]), result), 

637 "object": object_uri, 

638 "subjectClass": ctx.highest_priority_class, 

639 "subjectShape": ctx.highest_priority_shape, 

640 "objectShape": object_shape, 

641 } 

642 ctx.grouped_triples[display_name]["triples"].append(new_triple_data) 

643 

644 

645def _fetch_virtual_property_entities( 

646 reference_field: str, 

647 target_class: str | None, 

648 ctx: GroupingContext, 

649) -> list[str]: 

650 decoded_subject = unquote(str(ctx.subject)) 

651 

652 query = f""" 

653 SELECT DISTINCT ?entity 

654 WHERE {{ 

655 ?entity <{reference_field}> <{decoded_subject}> . 

656 """ 

657 

658 if target_class: 

659 query += f""" 

660 ?entity a <{target_class}> . 

661 """ 

662 

663 query += """ 

664 } 

665 """ 

666 

667 if ctx.historical_snapshot: 

668 return [ 

669 str(row[0]) for row in select_results(ctx.historical_snapshot.query(query)) 

670 ] 

671 

672 sparql = get_sparql() 

673 sparql.setQuery(query) 

674 sparql.setReturnFormat(JSON) 

675 bindings = get_sparql_bindings(sparql.query().convert()) 

676 return [res["entity"]["value"] for res in bindings] 

677 

678 

679def _build_virtual_property_triples( 

680 display_name: str, 

681 prop_config: dict, 

682 target_shape: str | None, 

683 entity_uris: list[str], 

684 ctx: GroupingContext, 

685) -> None: 

686 if display_name not in ctx.grouped_triples: 

687 ctx.grouped_triples[display_name] = { 

688 "property": display_name, 

689 "triples": [], 

690 "subjectClass": ctx.highest_priority_class, 

691 "subjectShape": ctx.highest_priority_shape, 

692 "objectShape": None, 

693 "is_virtual": True, 

694 } 

695 

696 for entity_uri in entity_uris: 

697 if ctx.historical_snapshot: 

698 result, external_entity = execute_historical_query( 

699 prop_config["fetchValueFromQuery"], 

700 ctx.subject, 

701 URIRef(entity_uri), 

702 ctx.historical_snapshot, 

703 ) 

704 else: 

705 result, external_entity = execute_sparql_query( 

706 prop_config["fetchValueFromQuery"], str(ctx.subject), entity_uri 

707 ) 

708 

709 if result: 

710 ctx.fetched_values_map[str(result)] = entity_uri 

711 new_triple_data = { 

712 "triple": (str(ctx.subject), display_name, str(result)), 

713 "external_entity": external_entity, 

714 "object": entity_uri, 

715 "subjectClass": ctx.highest_priority_class, 

716 "subjectShape": ctx.highest_priority_shape, 

717 "objectShape": target_shape, 

718 "is_virtual": True, 

719 } 

720 ctx.grouped_triples[display_name]["triples"].append(new_triple_data) 

721 

722 

723def process_virtual_property_display( 

724 display_name: str, 

725 prop_config: dict, 

726 ctx: GroupingContext, 

727) -> None: 

728 implementation = prop_config.get("implementedVia", {}) 

729 field_overrides = implementation.get("fieldOverrides", {}) 

730 target = implementation.get("target", {}) 

731 target_class = target.get("class") 

732 

733 reference_field = None 

734 for field_uri, override in field_overrides.items(): 

735 if override.get("value") == "${currentEntity}": 

736 reference_field = field_uri 

737 break 

738 

739 if not reference_field: 

740 return 

741 

742 entity_uris = _fetch_virtual_property_entities(reference_field, target_class, ctx) 

743 

744 if prop_config.get("fetchValueFromQuery") and entity_uris: 

745 _build_virtual_property_triples( 

746 display_name, prop_config, target.get("shape"), entity_uris, ctx 

747 ) 

748 

749 elif display_name not in ctx.grouped_triples: 

750 ctx.grouped_triples[display_name] = { 

751 "property": display_name, 

752 "triples": [], 

753 "subjectClass": ctx.highest_priority_class, 

754 "subjectShape": ctx.highest_priority_shape, 

755 "objectShape": None, 

756 "is_virtual": True, 

757 } 

758 

759 

760def execute_sparql_query( 

761 query: str, subject: str, value: str 

762) -> tuple[str | None, str | None]: 

763 sparql = get_sparql() 

764 

765 decoded_subject = unquote(subject) 

766 decoded_value = unquote(value) 

767 query = query.replace("[[subject]]", f"<{decoded_subject}>") 

768 query = query.replace("[[value]]", f"<{decoded_value}>") 

769 sparql.setQuery(query) 

770 sparql.setReturnFormat(JSON) 

771 bindings = get_sparql_bindings(sparql.query().convert()) 

772 if bindings: 

773 parsed_query = parseQuery(query) 

774 algebra_query = translateQuery(parsed_query).algebra 

775 variable_order = algebra_query["PV"] 

776 result = bindings[0] 

777 values = [ 

778 result.get(str(var_name), {}).get("value", None) 

779 for var_name in variable_order 

780 ] 

781 first_value = values[0] if len(values) > 0 else None 

782 second_value = values[1] if len(values) > 1 else None 

783 return (first_value, second_value) 

784 return None, None 

785 

786 

787def process_ordering( 

788 ctx: GroupingContext, 

789 prop: dict, 

790 order_property: str | None, 

791 display_name: str, 

792) -> None: 

793 def get_ordered_sequence( 

794 order_results: list[dict[str, dict[str, str]]] | list[ResultRow], 

795 ) -> list[list[str]]: 

796 order_map = {} 

797 for res in order_results: 

798 if isinstance(res, dict): # For live triplestore results 

799 ordered_entity = res["orderedEntity"]["value"] 

800 next_value = res["nextValue"]["value"] 

801 else: # For historical snapshot results 

802 ordered_entity = str(res[0]) 

803 next_value = str(res[1]) 

804 

805 order_map[str(ordered_entity)] = ( 

806 None if str(next_value) == "NONE" else str(next_value) 

807 ) 

808 

809 all_sequences = [] 

810 start_elements = set(order_map.keys()) - set(order_map.values()) 

811 while start_elements: 

812 sequence = [] 

813 current_element = start_elements.pop() 

814 while current_element in order_map: 

815 sequence.append(current_element) 

816 current_element = order_map[current_element] 

817 all_sequences.append(sequence) 

818 return all_sequences 

819 

820 decoded_subject = unquote(ctx.subject) 

821 

822 sparql = get_sparql() 

823 

824 order_query = f""" 

825 SELECT ?orderedEntity (COALESCE(?next, "NONE") AS ?nextValue) 

826 WHERE {{ 

827 <{decoded_subject}> <{prop["property"]}> ?orderedEntity. 

828 OPTIONAL {{ 

829 ?orderedEntity <{order_property}> ?next. 

830 }} 

831 }} 

832 """ 

833 if ctx.historical_snapshot: 

834 order_results: list[dict[str, dict[str, str]]] | list[ResultRow] = list( 

835 select_results(ctx.historical_snapshot.query(order_query)) 

836 ) 

837 else: 

838 sparql.setQuery(order_query) 

839 sparql.setReturnFormat(JSON) 

840 order_results = get_sparql_bindings(sparql.query().convert()) 

841 

842 order_sequences = get_ordered_sequence(order_results) 

843 for sequence in order_sequences: 

844 ctx.grouped_triples[display_name]["triples"].sort( 

845 key=lambda x: ( 

846 sequence.index( 

847 ctx.fetched_values_map.get(str(x["triple"][2]), str(x["triple"][2])) 

848 ) 

849 if ctx.fetched_values_map.get(str(x["triple"][2]), str(x["triple"][2])) 

850 in sequence 

851 else float("inf") 

852 ) 

853 ) 

854 

855 

856def process_default_property( 

857 prop_uri: str, 

858 triples: list[tuple[URIRef, URIRef, URIRef | Literal]], 

859 grouped_triples: OrderedDict, 

860 subject_shape: str | None = None, 

861 subject_class: str | None = None, 

862) -> None: 

863 display_name = prop_uri 

864 grouped_triples[display_name] = { 

865 "property": prop_uri, 

866 "triples": [], 

867 "subjectClass": subject_class,