Coverage for heritrace / sparql.py: 100%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5import logging 

6import time 

7from collections.abc import Iterator 

8from typing import TypedDict, cast 

9 

10from rdflib.query import Result, ResultRow 

11from SPARQLWrapper import POST, QueryResult, SPARQLWrapper 

12from SPARQLWrapper.SPARQLExceptions import SPARQLWrapperException 

13 

14 

15class SPARQLWrapperWithRetry(SPARQLWrapper): 

16 def __init__( 

17 self, 

18 endpoint: str, 

19 *, 

20 max_attempts: int = 3, 

21 initial_delay: float = 1.0, 

22 backoff_factor: float = 2.0, 

23 timeout: float = 5.0, 

24 ) -> None: 

25 self.max_attempts = max_attempts 

26 self.initial_delay = initial_delay 

27 self.backoff_factor = backoff_factor 

28 

29 super().__init__(endpoint) 

30 

31 self.setTimeout(int(timeout)) 

32 self.setMethod(POST) 

33 

34 def query(self) -> QueryResult: 

35 return self._query_with_retry() 

36 

37 def _query_with_retry(self) -> QueryResult: 

38 logger = logging.getLogger(__name__) 

39 

40 delay = self.initial_delay 

41 last_exception = None 

42 

43 for attempt in range(1, self.max_attempts + 1): 

44 try: 

45 return super().query() 

46 except (SPARQLWrapperException, OSError) as e: # noqa: PERF203 

47 last_exception = e 

48 logger.warning( 

49 "SPARQL query attempt %d/%d failed: %s", 

50 attempt, 

51 self.max_attempts, 

52 e, 

53 ) 

54 

55 if attempt < self.max_attempts: 

56 logger.info("Retrying in %.2f seconds...", delay) 

57 time.sleep(delay) 

58 delay *= self.backoff_factor 

59 

60 logger.error("All %d SPARQL query attempts failed", self.max_attempts) 

61 raise last_exception # type: ignore[misc] 

62 

63 

64class _SparqlJsonResults(TypedDict): 

65 bindings: list[dict[str, dict[str, str]]] 

66 

67 

68class _SparqlJsonResponse(TypedDict): 

69 results: _SparqlJsonResults 

70 

71 

72def get_sparql_bindings(result: object) -> list[dict[str, dict[str, str]]]: 

73 return cast("_SparqlJsonResponse", result)["results"]["bindings"] 

74 

75 

76def select_results(result: Result) -> Iterator[ResultRow]: 

77 for row in result: 

78 yield cast("ResultRow", row)