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 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 is_update: bool = False, 

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 if is_update: 

46 sparql.setMethod(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 max_retries: int = 5, 

75 backoff_factor: float = 0.5, 

76) -> SparqlQueryResult: 

77 raw = _execute_with_retry(endpoint, query, JSON, max_retries=max_retries, backoff_factor=backoff_factor) 

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

79 

80 

81def sparql_update( 

82 endpoint: str, 

83 query: str, 

84 *, 

85 max_retries: int = 5, 

86 backoff_factor: float = 0.5, 

87) -> None: 

88 _execute_with_retry(endpoint, query, JSON, is_update=True, max_retries=max_retries, backoff_factor=backoff_factor) 

89 

90 

91def sparql_construct( 

92 endpoint: str, 

93 query: str, 

94 *, 

95 max_retries: int = 5, 

96 backoff_factor: float = 0.5, 

97) -> bytes: 

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