Coverage for oc_ocdm / support / query_utils.py: 77%

1#!/usr/bin/python 

2 

3# SPDX-FileCopyrightText: 2020-2022 Simone Persiani <iosonopersia@gmail.com> 

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

5# 

6# SPDX-License-Identifier: ISC 

7 

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

9from __future__ import annotations 

10 

11from typing import TYPE_CHECKING, AbstractSet, List, Set 

12 

13from rdflib.term import Node 

14from triplelite import RDFTerm, Triple 

15 

16if TYPE_CHECKING: 

17 from typing import Tuple 

18 

19 from oc_ocdm.abstract_entity import AbstractEntity 

20 

21MAX_TRIPLES_PER_QUERY = 500 

22 

23 

24def _term_to_nt(term: str | RDFTerm | Node) -> str: 

25 if isinstance(term, RDFTerm): 

26 if term.type == "literal": 

27 escaped = term.value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n").replace("\r", "\\r") 

28 if term.lang: 

29 return f'"{escaped}"@{term.lang}' 

30 return f'"{escaped}"^^<{term.datatype}>' 

31 return f"<{term.value}>" 

32 if isinstance(term, str): 

33 return f"<{term}>" 

34 return term.n3() 

35 

36 

37def _serialize_triples_to_nt(triples: AbstractSet[Triple]) -> str: 

38 return "".join(f"{_term_to_nt(s)} {_term_to_nt(p)} {_term_to_nt(o)} ." for s, p, o in triples) 

39 

40 

41def _chunk_set(data: AbstractSet[Triple], chunk_size: int) -> List[Set[Triple]]: 

42 data_list = list(data) 

43 return [set(data_list[i : i + chunk_size]) for i in range(0, len(data_list), chunk_size)] 

44 

45 

46def get_delete_query(graph_iri: str, data: AbstractSet[Triple]) -> Tuple[List[str], int]: 

47 num_of_statements: int = len(data) 

48 if num_of_statements <= 0: 

49 return [], 0 

50 

51 if num_of_statements <= MAX_TRIPLES_PER_QUERY: 

52 statements: str = _serialize_triples_to_nt(data) 

53 return [f"DELETE DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}"], num_of_statements 

54 

55 chunks = _chunk_set(data, MAX_TRIPLES_PER_QUERY) 

56 queries: List[str] = [] 

57 for chunk in chunks: 

58 statements = _serialize_triples_to_nt(chunk) 

59 queries.append(f"DELETE DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}") 

60 return queries, num_of_statements 

61 

62 

63def get_insert_query(graph_iri: str, data: AbstractSet[Triple]) -> Tuple[List[str], int]: 

64 num_of_statements: int = len(data) 

65 if num_of_statements <= 0: 

66 return [], 0 

67 

68 if num_of_statements <= MAX_TRIPLES_PER_QUERY: 

69 statements: str = _serialize_triples_to_nt(data) 

70 return [f"INSERT DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}"], num_of_statements 

71 

72 chunks = _chunk_set(data, MAX_TRIPLES_PER_QUERY) 

73 queries: List[str] = [] 

74 for chunk in chunks: 

75 statements = _serialize_triples_to_nt(chunk) 

76 queries.append(f"INSERT DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}") 

77 return queries, num_of_statements 

78 

79 

80def _compute_graph_changes(entity: AbstractEntity, entity_type: str) -> Tuple[Set[Triple], Set[Triple], int, int]: 

81 """ 

82 Computes the triples to insert and delete for an entity. 

83 

84 Args: 

85 entity: The entity to analyze 

86 entity_type: Type of entity ("graph", "prov", or "metadata") 

87 

88 Returns: 

89 Tuple of (triples_to_insert, triples_to_delete, added_count, removed_count) 

90 """ 

91 if entity_type == "prov": 

92 triples = set(entity.g) 

93 return triples, set(), len(triples), 0 

94 

95 # Deferred import to break circular dependency: 

96 # graph_entity → abstract_entity → support.support → (support/__init__) → query_utils → graph_entity 

97 from oc_ocdm.graph.graph_entity import GraphEntity # noqa: E402 

98 

99 assert isinstance(entity, GraphEntity) 

100 to_be_deleted: bool = entity.to_be_deleted 

101 preexisting_triples = entity.preexisting_triples 

102 

103 if to_be_deleted: 

104 return set(), set(preexisting_triples), 0, len(preexisting_triples) 

105 

106 current_triples = set(entity.g) 

107 

108 if len(preexisting_triples) == len(current_triples) and set(preexisting_triples) == current_triples: 

109 return set(), set(), 0, 0 

110 

111 removed_triples = set(preexisting_triples) - current_triples 

112 added_triples = current_triples - set(preexisting_triples) 

113 

114 return added_triples, removed_triples, len(added_triples), len(removed_triples) 

115 

116 

117def get_update_query(entity: AbstractEntity, entity_type: str = "graph") -> Tuple[List[str], int, int]: 

118 to_insert, to_delete, n_added, n_removed = _compute_graph_changes(entity, entity_type) 

119 

120 if n_added == 0 and n_removed == 0: 

121 return [], 0, 0 

122 

123 graph_iri = entity.g.identifier 

124 if graph_iri is None: 

125 raise ValueError("Entity graph has no identifier") 

126 

127 delete_queries, _ = get_delete_query(graph_iri, to_delete) 

128 insert_queries, _ = get_insert_query(graph_iri, to_insert) 

129 

130 return delete_queries + insert_queries, n_added, n_removed