Coverage for oc_ocdm / support / sparql.py: 72%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import json 

8import time 

9from typing import cast 

10from urllib.error import HTTPError, URLError 

11from urllib.parse import parse_qs, urlparse 

12 

13from SPARQLWrapper import GET, JSON, N3, POST, URLENCODED, SPARQLWrapper 

14 

15from oc_ocdm._types import SparqlQueryResult 

16 

17 

18class SPARQLEndpointError(Exception): 

19 def __init__(self, message: str, status_code: int | None = None): 

20 super().__init__(message) 

21 self.status_code = status_code 

22 

23 

24def _make_sparql_client(endpoint: str) -> SPARQLWrapper: 

25 parsed = urlparse(endpoint) 

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

27 sparql = SPARQLWrapper(base_url) 

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

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

30 return sparql 

31 

32 

33def _execute_with_retry( 

34 endpoint: str, 

35 query: str, 

36 return_format: str, 

37 *, 

38 method: str = GET, 

39 max_retries: int = 5, 

40 backoff_factor: float = 0.5, 

41) -> bytes: 

42 sparql = _make_sparql_client(endpoint) 

43 sparql.setQuery(query) 

44 sparql.setReturnFormat(return_format) 

45 sparql.setMethod(method) 

46 if method == POST: 

47 sparql.setRequestMethod(URLENCODED) 

48 

49 last_error: SPARQLEndpointError | None = None 

50 

51 for attempt in range(max_retries + 1): 

52 if attempt > 0: 

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

54 try: 

55 return sparql.query().response.read() 

56 except HTTPError as e: 

57 if e.code == 400: 

58 raise SPARQLEndpointError(f"Query syntax error: {e.read().decode()}", status_code=400) from e 

59 if e.code >= 500: 

60 last_error = SPARQLEndpointError(f"Server error: {e.code}", status_code=e.code) 

61 continue 

62 raise SPARQLEndpointError(f"HTTP error: {e.code} - {e.read().decode()}", status_code=e.code) from e 

63 except URLError as e: 

64 last_error = SPARQLEndpointError(f"Connection error: {e.reason}") 

65 continue 

66 

67 raise last_error # type: ignore[misc] 

68 

69 

70def sparql_query( 

71 endpoint: str, 

72 query: str, 

73 *, 

74 method: str = GET, 

75 max_retries: int = 5, 

76 backoff_factor: float = 0.5, 

77) -> SparqlQueryResult: 

78 raw = _execute_with_retry( 

79 endpoint, 

80 query, 

81 JSON, 

82 method=method, 

83 max_retries=max_retries, 

84 backoff_factor=backoff_factor, 

85 ) 

86 return cast(SparqlQueryResult, json.loads(raw)) 

87 

88 

89def sparql_update( 

90 endpoint: str, 

91 query: str, 

92 *, 

93 max_retries: int = 5, 

94 backoff_factor: float = 0.5, 

95) -> None: 

96 _execute_with_retry(endpoint, query, JSON, method=POST, max_retries=max_retries, backoff_factor=backoff_factor) 

97 

98 

99def sparql_construct( 

100 endpoint: str, 

101 query: str, 

102 *, 

103 max_retries: int = 5, 

104 backoff_factor: float = 0.5, 

105) -> bytes: 

106 return _execute_with_retry(endpoint, query, N3, max_retries=max_retries, backoff_factor=backoff_factor)