Coverage for rdflib_ocdm / query_utils.py: 94%

1#!/usr/bin/python 

2 

3# SPDX-FileCopyrightText: 2016 Silvio Peroni <essepuntato@gmail.com> 

4# SPDX-FileCopyrightText: 2023-2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

5# 

6# SPDX-License-Identifier: ISC 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING, cast 

11 

12if TYPE_CHECKING: 

13 from typing import Tuple 

14 

15 from rdflib.compare import IsomorphicGraph 

16 

17 from rdflib_ocdm.ocdm_graph import OCDMGraphCommons 

18 

19from rdflib import Dataset, Graph, URIRef 

20from rdflib.compare import graph_diff, to_isomorphic 

21 

22from rdflib_ocdm.graph_utils import _extract_graph_iri 

23from rdflib_ocdm.support import get_entity_subgraph 

24 

25 

26def get_delete_query( 

27 data: Dataset | Graph, graph_iri: URIRef | None = None 

28) -> Tuple[str, int]: 

29 num_of_statements: int = len(data) 

30 if num_of_statements <= 0: 

31 return "", 0 

32 else: 

33 statements: str = data.serialize(format="nt11").replace("\n", "") 

34 if graph_iri: 

35 delete_string: str = ( 

36 f"DELETE DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}" 

37 ) 

38 else: 

39 delete_string: str = f"DELETE DATA {{ {statements} }}" 

40 return delete_string, num_of_statements 

41 

42 

43def get_insert_query( 

44 data: Dataset | Graph, graph_iri: URIRef | None = None 

45) -> Tuple[str, int]: 

46 num_of_statements: int = len(data) 

47 if num_of_statements <= 0: 

48 return "", 0 

49 else: 

50 statements: str = data.serialize(format="nt11").replace("\n", "") 

51 if graph_iri: 

52 insert_string: str = ( 

53 f"INSERT DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}" 

54 ) 

55 else: 

56 insert_string: str = f"INSERT DATA {{ {statements} }}" 

57 return insert_string, num_of_statements 

58 

59 

60def get_update_query( 

61 a_set: OCDMGraphCommons | Dataset | Graph, 

62 entity: URIRef, 

63 entity_type: str = "graph", 

64) -> Tuple[str, int, int]: 

65 to_be_deleted: bool = False 

66 preexisting_graph: Dataset | Graph = Dataset() 

67 graph_iri: URIRef | None = None 

68 

69 if entity_type == "graph": 

70 assert hasattr(a_set, "entity_index") and hasattr(a_set, "preexisting_graph") 

71 graph_set = cast("OCDMGraphCommons", a_set) 

72 to_be_deleted = ( 

73 graph_set.entity_index[entity]["to_be_deleted"] 

74 if entity in graph_set.entity_index 

75 else False 

76 ) 

77 preexisting_graph = get_entity_subgraph(graph_set.preexisting_graph, entity) 

78 

79 if isinstance(a_set, Dataset): 

80 if hasattr(a_set, "entity_index") and entity in a_set.entity_index: # type: ignore[operator] 

81 graph_iri = a_set.entity_index[entity].get("graph_iri") # type: ignore[union-attr] 

82 else: 

83 graph_iri = _extract_graph_iri(a_set, entity) 

84 

85 if to_be_deleted: 

86 delete_string, removed_triples = get_delete_query(preexisting_graph, graph_iri) 

87 if delete_string != "": 87 ↛ 90line 87 didn't jump to line 90 because the condition on line 87 was always true

88 return delete_string, 0, removed_triples 

89 else: 

90 return "", 0, 0 

91 else: 

92 assert isinstance(a_set, (Graph, Dataset)) 

93 current_graph = get_entity_subgraph(a_set, entity) 

94 

95 # Convert Dataset to Graph for isomorphic comparison if needed 

96 if isinstance(preexisting_graph, Dataset): 

97 preexisting_for_comparison = Graph() 

98 for s, p, o, _ in preexisting_graph.quads((None, None, None, None)): 

99 preexisting_for_comparison.add((s, p, o)) 

100 else: 

101 preexisting_for_comparison = preexisting_graph 

102 

103 if isinstance(current_graph, Dataset): 

104 current_for_comparison = Graph() 

105 for s, p, o, _ in current_graph.quads((None, None, None, None)): 

106 current_for_comparison.add((s, p, o)) 

107 else: 

108 current_for_comparison = current_graph 

109 

110 preexisting_iso: IsomorphicGraph = to_isomorphic(preexisting_for_comparison) 

111 current_iso: IsomorphicGraph = to_isomorphic(current_for_comparison) 

112 if preexisting_iso == current_iso: 

113 # Both graphs have exactly the same content! 

114 return "", 0, 0 

115 _, in_first, in_second = graph_diff(preexisting_iso, current_iso) 

116 delete_string, removed_triples = get_delete_query(in_first, graph_iri) 

117 insert_string, added_triples = get_insert_query(in_second, graph_iri) 

118 if delete_string != "" and insert_string != "": 

119 return delete_string + "; " + insert_string, added_triples, removed_triples 

120 elif delete_string != "": 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true

121 return delete_string, 0, removed_triples 

122 elif insert_string != "": 122 ↛ 125line 122 didn't jump to line 125 because the condition on line 122 was always true

123 return insert_string, added_triples, 0 

124 else: 

125 return "", 0, 0