Coverage for rdflib_ocdm/query_utils.py: 89%

1#!/usr/bin/python 

2# -*- coding: utf-8 -*- 

3# Copyright (c) 2016, Silvio Peroni <essepuntato@gmail.com> 

4# 

5# Permission to use, copy, modify, and/or distribute this software for any purpose 

6# with or without fee is hereby granted, provided that the above copyright notice 

7# and this permission notice appear in all copies. 

8# 

9# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH 

10# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND 

11# FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, 

12# OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, 

13# DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS 

14# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS 

15# SOFTWARE. 

16 

17from __future__ import annotations 

18 

19from typing import TYPE_CHECKING 

20 

21if TYPE_CHECKING: 21 ↛ 22line 21 didn't jump to line 22 because the condition on line 21 was never true

22 from typing import Tuple 

23 from rdflib.compare import IsomorphicGraph 

24 from rdflib_ocdm.ocdm_graph import OCDMGraphCommons 

25 

26from rdflib import Dataset, Graph, URIRef 

27from rdflib.compare import graph_diff, to_isomorphic 

28 

29from rdflib_ocdm.graph_utils import _extract_graph_iri 

30from rdflib_ocdm.support import get_entity_subgraph 

31 

32 

33def get_delete_query(data: Dataset|Graph, graph_iri: URIRef = None) -> Tuple[str, int]: 

34 num_of_statements: int = len(data) 

35 if num_of_statements <= 0: 

36 return "", 0 

37 else: 

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

39 if graph_iri: 

40 delete_string: str = f"DELETE DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}" 

41 else: 

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

43 return delete_string, num_of_statements 

44 

45def get_insert_query(data: Dataset|Graph, graph_iri: URIRef = None) -> 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 = f"INSERT DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}" 

53 else: 

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

55 return insert_string, num_of_statements 

56 

57def get_update_query(a_set: OCDMGraphCommons|Dataset|Graph, entity: URIRef, entity_type = 'graph') -> Tuple[str, int, int]: 

58 if entity_type == 'graph': 

59 to_be_deleted: bool = a_set.entity_index[entity]['to_be_deleted'] if entity in a_set.entity_index else False 

60 preexisting_graph = get_entity_subgraph(a_set.preexisting_graph, entity) 

61 graph_iri = None 

62 elif entity_type == 'prov': 62 ↛ 67line 62 didn't jump to line 67 because the condition on line 62 was always true

63 to_be_deleted = False 

64 preexisting_graph = Dataset() 

65 

66 # Extract graph_iri: prefer entity_index (OCDMDataset), fallback to helper function (regular Dataset) 

67 if isinstance(a_set, Dataset): 

68 if hasattr(a_set, 'entity_index') and entity in a_set.entity_index: 

69 # Clean architectural solution: use stored graph_iri from entity_index 

70 graph_iri = a_set.entity_index[entity].get('graph_iri') 

71 else: 

72 # Fallback for regular Dataset (e.g., provenance graphs): use DRY helper function 

73 graph_iri = _extract_graph_iri(a_set, entity) 

74 elif isinstance(a_set, Graph): 74 ↛ 76line 74 didn't jump to line 76 because the condition on line 74 was always true

75 graph_iri = None 

76 if to_be_deleted: 

77 delete_string, removed_triples = get_delete_query(preexisting_graph, graph_iri) 

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

79 return delete_string, 0, removed_triples 

80 else: 

81 return "", 0, 0 

82 else: 

83 current_graph = get_entity_subgraph(a_set, entity) 

84 

85 # Convert Dataset to Graph for isomorphic comparison if needed 

86 if isinstance(preexisting_graph, Dataset): 

87 preexisting_for_comparison = Graph() 

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

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

90 else: 

91 preexisting_for_comparison = preexisting_graph 

92 

93 if isinstance(current_graph, Dataset): 

94 current_for_comparison = Graph() 

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

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

97 else: 

98 current_for_comparison = current_graph 

99 

100 preexisting_iso: IsomorphicGraph = to_isomorphic(preexisting_for_comparison) 

101 current_iso: IsomorphicGraph = to_isomorphic(current_for_comparison) 

102 if preexisting_iso == current_iso: 

103 # Both graphs have exactly the same content! 

104 return "", 0, 0 

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

106 delete_string, removed_triples = get_delete_query(in_first, graph_iri) 

107 insert_string, added_triples = get_insert_query(in_second, graph_iri) 

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

109 return delete_string + '; ' + insert_string, added_triples, removed_triples 

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

111 return delete_string, 0, removed_triples 

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

113 return insert_string, added_triples, 0 

114 else: 

115 return "", 0, 0