Coverage for src / time_agnostic_library / sparql.py: 100%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5 

6import atexit 

7import threading 

8import zipfile 

9 

10from rdflib import Dataset 

11from rdflib.term import Literal, URIRef 

12from sparqlite import SPARQLClient 

13 

14from time_agnostic_library.prov_entity import ProvEntity 

15 

16__all__ = [ 

17 "Sparql", 

18 "_binding_to_n3", 

19 "_n3_to_binding", 

20 "_n3_value", 

21] 

22 

23CONFIG_PATH = "./config.json" 

24 

25_PROV_PROPERTY_STRINGS: tuple[str, ...] = tuple(ProvEntity.get_prov_properties()) 

26 

27_client_cache: dict[tuple[str, int], SPARQLClient] = {} 

28_client_lock = threading.Lock() 

29 

30 

31def _get_client(url: str) -> SPARQLClient: 

32 key = (url, threading.get_ident()) 

33 with _client_lock: 

34 client = _client_cache.get(key) 

35 if client is None: 

36 client = SPARQLClient(url) 

37 _client_cache[key] = client 

38 return client 

39 

40 

41def _close_all_clients() -> None: 

42 with _client_lock: 

43 for client in _client_cache.values(): 

44 client.close() 

45 _client_cache.clear() 

46 

47 

48atexit.register(_close_all_clients) 

49 

50 

51def _escape_n3(v: str) -> str: 

52 return ( 

53 v.replace("\\", "\\\\") 

54 .replace('"', '\\"') 

55 .replace("\n", "\\n") 

56 .replace("\r", "\\r") 

57 ) 

58 

59 

60def _binding_to_n3(val: dict) -> str: 

61 if val["type"] == "uri": 

62 return f"<{val['value']}>" 

63 if val["type"] == "bnode": 

64 return f"_:{val['value']}" 

65 escaped = _escape_n3(val["value"]) 

66 if "datatype" in val: 

67 return f'"{escaped}"^^<{val["datatype"]}>' 

68 if "xml:lang" in val: 

69 return f'"{escaped}"@{val["xml:lang"]}' 

70 return f'"{escaped}"' 

71 

72 

73def _find_closing_quote(n3: str) -> int: 

74 pos = n3.find('"', 1) 

75 while pos > 0: 

76 num_backslashes = 0 

77 check = pos - 1 

78 while check >= 1 and n3[check] == "\\": 

79 num_backslashes += 1 

80 check -= 1 

81 if num_backslashes % 2 == 0: 

82 return pos 

83 pos = n3.find('"', pos + 1) 

84 return -1 

85 

86 

87def _unescape_n3(raw: str) -> str: 

88 out: list[str] = [] 

89 i = 0 

90 while i < len(raw): 

91 if raw[i] == "\\" and i + 1 < len(raw): 

92 nxt = raw[i + 1] 

93 if nxt == "n": 

94 out.append("\n") 

95 elif nxt == "r": 

96 out.append("\r") 

97 elif nxt == '"': 

98 out.append('"') 

99 elif nxt == "\\": 

100 out.append("\\") 

101 else: 

102 out.append(raw[i]) 

103 out.append(nxt) 

104 i += 2 

105 else: 

106 out.append(raw[i]) 

107 i += 1 

108 return "".join(out) 

109 

110 

111def _parse_n3_literal(n3: str) -> tuple[str, str]: 

112 quote_end = _find_closing_quote(n3) 

113 if quote_end == -1: 

114 return n3, "" 

115 raw = n3[1:quote_end] 

116 return _unescape_n3(raw), n3[quote_end + 1 :] 

117 

118 

119def _n3_value(n3: str) -> str: 

120 if n3.startswith("<") and n3.endswith(">"): 

121 return n3[1:-1] 

122 if n3.startswith("_:"): 

123 return n3[2:] 

124 value, _ = _parse_n3_literal(n3) 

125 return value 

126 

127 

128def _n3_to_binding(n3: str) -> dict: 

129 if n3.startswith("<") and n3.endswith(">"): 

130 return {"type": "uri", "value": n3[1:-1]} 

131 if n3.startswith("_:"): 

132 return {"type": "bnode", "value": n3[2:]} 

133 value, rest = _parse_n3_literal(n3) 

134 if rest.startswith("^^<") and rest.endswith(">"): 

135 return {"type": "literal", "value": value, "datatype": rest[3:-1]} 

136 if rest.startswith("@"): 

137 return {"type": "literal", "value": value, "xml:lang": rest[1:]} 

138 return {"type": "literal", "value": value} 

139 

140 

141class Sparql: 

142 def __init__(self, query: str, config: dict): 

143 self.query = query 

144 self.config = config 

145 if any(uri in query for uri in _PROV_PROPERTY_STRINGS): 

146 self.storer: dict = config["provenance"] 

147 else: 

148 self.storer: dict = config["dataset"] 

149 

150 def run_select_query(self) -> dict: 

151 output = {"head": {"vars": []}, "results": {"bindings": []}} 

152 if self.storer["file_paths"]: 

153 output = self._get_results_from_files(output) 

154 if self.storer["triplestore_urls"]: 

155 output = self._get_results_from_triplestores(output) 

156 return output 

157 

158 def _get_results_from_files(self, output: dict) -> dict: 

159 storer: list[str] = self.storer["file_paths"] 

160 for file_path in storer: 

161 file_cg = Dataset(default_union=True) 

162 if file_path.endswith(".zip"): 

163 with ( 

164 zipfile.ZipFile(file_path, "r") as z, 

165 z.open(z.namelist()[0]) as file, 

166 ): 

167 file_cg.parse(file=file, format="json-ld") # type: ignore[arg-type] 

168 else: 

169 file_cg.parse(location=file_path, format="json-ld") 

170 query_results = file_cg.query(self.query) 

171 vars_list = [str(var) for var in query_results.vars or []] 

172 output["head"]["vars"] = vars_list 

173 for result in query_results: 

174 binding = {} 

175 for var in vars_list: 

176 value = result[var] # type: ignore[index] 

177 if value is not None: 

178 binding[var] = self._format_result_value(value) 

179 output["results"]["bindings"].append(binding) 

180 return output 

181 

182 def _get_results_from_triplestores(self, output: dict) -> dict: 

183 storer = self.storer["triplestore_urls"] 

184 for url in storer: 

185 results = _get_client(url).query(self.query) 

186 if not output["head"]["vars"]: 

187 output["head"]["vars"] = results["head"]["vars"] 

188 output["results"]["bindings"].extend(results["results"]["bindings"]) 

189 return output 

190 

191 @staticmethod 

192 def _format_result_value(value) -> dict: 

193 if isinstance(value, URIRef): 

194 return {"type": "uri", "value": str(value)} 

195 if isinstance(value, Literal): 

196 result = {"type": "literal", "value": str(value)} 

197 if value.datatype: 

198 result["datatype"] = str(value.datatype) 

199 if value.language: 

200 result["xml:lang"] = value.language 

201 return result 

202 return {"type": "literal", "value": str(value)} 

203 

204 def run_select_to_quad_set(self) -> set[tuple[str, ...]]: 

205 results = self.run_select_query() 

206 output: set[tuple[str, ...]] = set() 

207 vars_list = results["head"]["vars"] 

208 for binding in results["results"]["bindings"]: 

209 components: list[str] = [] 

210 skip = False 

211 for var in vars_list: 

212 if var not in binding: 

213 skip = True 

214 break 

215 components.append(_binding_to_n3(binding[var])) 

216 if not skip: 

217 output.add(tuple(components)) 

218 return output 

219 

220 def run_ask_query(self) -> bool: 

221 storer = self.storer["triplestore_urls"] 

222 for url in storer: 

223 return _get_client(url).ask(self.query) 

224 return False 

225 

226 @classmethod 

227 def _get_tuples_set(cls, result_dict: dict, output: set, vars_list: list) -> None: 

228 results_list = [] 

229 for var in vars_list: 

230 if str(var) in result_dict: 

231 val = result_dict[str(var)] 

232 if isinstance(val, dict) and "value" in val: 

233 results_list.append(str(val["value"])) 

234 else: 

235 results_list.append(str(val)) 

236 else: 

237 results_list.append(None) 

238 output.add(tuple(results_list))