Coverage for oc_meta / lib / sparql.py: 75%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import multiprocessing 

8import time 

9from concurrent.futures import ProcessPoolExecutor, as_completed 

10from typing import Callable 

11from urllib.error import URLError 

12from urllib.parse import parse_qs, urlparse 

13 

14from SPARQLWrapper import GET, JSON, POST, URLENCODED, SPARQLWrapper 

15from SPARQLWrapper.SPARQLExceptions import EndPointInternalError, QueryBadFormed 

16 

17from oc_meta.constants import QLEVER_MAX_WORKERS, QLEVER_QUERIES_PER_GROUP 

18 

19 

20def _make_sparql_client(endpoint_url: str, timeout: int = 3600) -> SPARQLWrapper: 

21 parsed = urlparse(endpoint_url) 

22 base_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}" 

23 sparql = SPARQLWrapper(base_url) 

24 for key, values in parse_qs(parsed.query).items(): 

25 sparql.addParameter(key, values[0]) 

26 sparql.setReturnFormat(JSON) 

27 sparql.setTimeout(timeout) 

28 return sparql 

29 

30 

31def execute_sparql( 

32 endpoint_url: str, 

33 query: str, 

34 max_retries: int = 5, 

35 backoff_factor: float = 5, 

36 timeout: int = 3600, 

37 *, 

38 method: str = GET, 

39) -> dict: 

40 sparql = _make_sparql_client(endpoint_url, timeout) 

41 sparql.setMethod(method) 

42 if method == POST: 

43 sparql.setRequestMethod(URLENCODED) 

44 last_error: Exception | None = None 

45 for attempt in range(max_retries + 1): 

46 if attempt > 0: 

47 time.sleep(backoff_factor * (2**attempt)) 

48 try: 

49 sparql.setQuery(query) 

50 return sparql.queryAndConvert() # type: ignore[return-value] 

51 except QueryBadFormed: 

52 raise 

53 except (EndPointInternalError, URLError) as e: 

54 last_error = e 

55 raise last_error # type: ignore[misc] 

56 

57 

58def execute_sparql_update( 

59 endpoint_url: str, 

60 query: str, 

61 max_retries: int = 5, 

62 backoff_factor: float = 5, 

63 timeout: int = 3600, 

64) -> None: 

65 sparql = _make_sparql_client(endpoint_url, timeout) 

66 sparql.setMethod(POST) 

67 last_error: Exception | None = None 

68 for attempt in range(max_retries + 1): 

69 if attempt > 0: 

70 time.sleep(backoff_factor * (2**attempt)) 

71 try: 

72 sparql.setQuery(query) 

73 sparql.query() 

74 return 

75 except QueryBadFormed: 

76 raise 

77 except (EndPointInternalError, URLError) as e: 

78 last_error = e 

79 raise last_error # type: ignore[misc] 

80 

81 

82def execute_sparql_queries( 

83 endpoint_url: str, 

84 queries: list[str], 

85 max_retries: int = 5, 

86 backoff_factor: float = 5, 

87 timeout: int = 3600, 

88) -> list[list[dict[str, dict[str, str]]]]: 

89 results: list[list[dict[str, dict[str, str]]]] = [] 

90 sparql = _make_sparql_client(endpoint_url, timeout) 

91 for query in queries: 

92 last_error: Exception | None = None 

93 for attempt in range(max_retries + 1): 

94 if attempt > 0: 

95 time.sleep(backoff_factor * (2**attempt)) 

96 try: 

97 sparql.setQuery(query) 

98 result: dict[str, dict[str, list[dict[str, dict[str, str]]]]] = ( 

99 sparql.queryAndConvert() 

100 ) # type: ignore[assignment] 

101 results.append(result["results"]["bindings"]) 

102 break 

103 except QueryBadFormed: 

104 raise 

105 except (EndPointInternalError, URLError) as e: 

106 last_error = e 

107 else: 

108 raise last_error # type: ignore[misc] 

109 return results 

110 

111 

112def run_queries_parallel( 

113 endpoint_url: str, 

114 batch_queries: list[str], 

115 batch_sizes: list[int], 

116 workers: int = QLEVER_MAX_WORKERS, 

117 progress_callback: Callable[[int], None] | None = None, 

118 max_retries: int = 5, 

119 backoff_factor: int = 5, 

120 timeout: int = 3600, 

121) -> list[list]: 

122 if not batch_queries: 

123 return [] 

124 

125 all_bindings: list[list] = [] 

126 

127 if len(batch_queries) > 1 and workers > 1: 

128 query_groups: list[list[str]] = [] 

129 grouped_sizes: list[int] = [] 

130 for i in range(0, len(batch_queries), QLEVER_QUERIES_PER_GROUP): 

131 query_groups.append(batch_queries[i : i + QLEVER_QUERIES_PER_GROUP]) 

132 grouped_sizes.append(sum(batch_sizes[i : i + QLEVER_QUERIES_PER_GROUP])) 

133 

134 with ProcessPoolExecutor( 

135 max_workers=min(len(query_groups), workers), 

136 mp_context=multiprocessing.get_context("forkserver"), 

137 ) as executor: 

138 future_to_size = { 

139 executor.submit( 

140 execute_sparql_queries, 

141 endpoint_url=endpoint_url, 

142 queries=group, 

143 max_retries=max_retries, 

144 backoff_factor=backoff_factor, 

145 timeout=timeout, 

146 ): size 

147 for group, size in zip(query_groups, grouped_sizes) 

148 } 

149 for future in as_completed(future_to_size): 

150 all_bindings.extend(future.result()) 

151 if progress_callback: 

152 progress_callback(future_to_size[future]) 

153 else: 

154 results = execute_sparql_queries( 

155 endpoint_url=endpoint_url, 

156 queries=batch_queries, 

157 max_retries=max_retries, 

158 backoff_factor=backoff_factor, 

159 timeout=timeout, 

160 ) 

161 all_bindings.extend(results) 

162 if progress_callback: 

163 progress_callback(sum(batch_sizes)) 

164 

165 return all_bindings