Coverage for rdflib_ocdm / storer.py: 96%

1#!/usr/bin/python 

2 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7from __future__ import annotations 

8 

9import os 

10from datetime import datetime 

11from typing import TYPE_CHECKING 

12 

13from oc_ocdm.support.reporter import Reporter 

14from SPARQLWrapper import SPARQLWrapper 

15 

16from rdflib_ocdm.ocdm_graph import OCDMDataset, OCDMGraph, OCDMGraphCommons 

17from rdflib_ocdm.query_utils import get_update_query 

18from rdflib_ocdm.retry_utils import execute_with_retry 

19 

20if TYPE_CHECKING: 

21 from typing import Set 

22 

23 from rdflib import Graph 

24 

25 

26class Storer: 

27 def __init__( 

28 self, 

29 abstract_set: OCDMGraphCommons | Graph, 

30 repok: Reporter | None = None, 

31 reperr: Reporter | None = None, 

32 output_format: str = "json-ld", 

33 zip_output: bool = False, 

34 ) -> None: 

35 self.a_set = abstract_set 

36 supported_formats: Set[str] = { 

37 "application/n-triples", 

38 "ntriples", 

39 "nt", 

40 "nt11", 

41 "application/n-quads", 

42 "nquads", 

43 "json-ld", 

44 } 

45 if output_format not in supported_formats: 

46 raise ValueError( 

47 f"Given output_format '{output_format}' is not supported." 

48 f" Available formats: {supported_formats}." 

49 ) 

50 else: 

51 self.output_format: str = output_format 

52 self.zip_output = zip_output 

53 if repok is None: 

54 self.repok: Reporter = Reporter(prefix="[Storer: INFO] ") 

55 else: 

56 self.repok: Reporter = repok 

57 

58 if reperr is None: 

59 self.reperr: Reporter = Reporter(prefix="[Storer: ERROR] ") 

60 else: 

61 self.reperr: Reporter = reperr 

62 

63 def _query( 

64 self, 

65 query_string: str, 

66 triplestore_url: str, 

67 base_dir: str | None = None, 

68 added_statements: int = 0, 

69 removed_statements: int = 0, 

70 max_retries: int = 5, 

71 ) -> bool: 

72 if query_string != "": 72 ↛ 112line 72 didn't jump to line 112 because the condition on line 72 was always true

73 try: 

74 # Use the retry utility function with custom error handling 

75 def execute_query(): 

76 sparql: SPARQLWrapper = SPARQLWrapper(triplestore_url) 

77 sparql.setQuery(query_string) 

78 sparql.setMethod("POST") 

79 sparql.query() 

80 return True 

81 

82 execute_with_retry( 

83 execute_query, max_retries=max_retries, reporter=self.repok 

84 ) 

85 

86 self.repok.add_sentence( 

87 f"Triplestore updated with {added_statements} added statements and " 

88 f"with {removed_statements} removed statements." 

89 ) 

90 

91 return True 

92 except ValueError as e: 

93 # Handle the case when all retries failed 

94 self.reperr.add_sentence( 

95 "[3] Graph was not loaded into the triplestore" 

96 f" due to communication problems: {e}" 

97 ) 

98 if base_dir is not None: 

99 tp_err_dir: str = base_dir + os.sep + "tp_err" 

100 if not os.path.exists(tp_err_dir): 100 ↛ 102line 100 didn't jump to line 102 because the condition on line 100 was always true

101 os.makedirs(tp_err_dir) 

102 cur_file_err: str = ( 

103 tp_err_dir 

104 + os.sep 

105 + datetime.now().strftime( 

106 "%Y-%m-%d-%H-%M-%S-%f_not_uploaded.txt" 

107 ) 

108 ) 

109 with open(cur_file_err, "wt", encoding="utf-8") as f: 

110 f.write(query_string) 

111 return False 

112 return False 

113 

114 def upload_all( 

115 self, triplestore_url: str, base_dir: str | None = None, batch_size: int = 10 

116 ) -> bool: 

117 self.repok.new_article() 

118 self.reperr.new_article() 

119 if batch_size <= 0: 

120 batch_size = 10 

121 query_string: str = "" 

122 added_statements: int = 0 

123 removed_statements: int = 0 

124 skipped_queries: int = 0 

125 result: bool = True 

126 entity_type = ( 

127 "graph" if isinstance(self.a_set, (OCDMGraph, OCDMDataset)) else "prov" 

128 ) 

129 for idx, entity in enumerate(list(self.a_set.all_entities)): # type: ignore[union-attr] 

130 update_query, n_added, n_removed = get_update_query( 

131 self.a_set, entity, entity_type 

132 ) 

133 if update_query == "": 

134 skipped_queries += 1 

135 else: 

136 index = idx - skipped_queries 

137 if index == 0: 

138 # First query 

139 query_string = update_query 

140 added_statements = n_added 

141 removed_statements = n_removed 

142 elif index % batch_size == 0: 

143 # batch_size-multiple query 

144 result &= self._query( 

145 query_string, 

146 triplestore_url, 

147 base_dir, 

148 added_statements, 

149 removed_statements, 

150 ) 

151 query_string = update_query 

152 added_statements = n_added 

153 removed_statements = n_removed 

154 else: 

155 # Accumulated query 

156 query_string += " ; " + update_query 

157 added_statements += n_added 

158 removed_statements += n_removed 

159 if query_string != "": 159 ↛ 167line 159 didn't jump to line 167 because the condition on line 159 was always true

160 result &= self._query( 

161 query_string, 

162 triplestore_url, 

163 base_dir, 

164 added_statements, 

165 removed_statements, 

166 ) 

167 return result