Coverage for rdflib_ocdm / reader.py: 100%

1#!/usr/bin/python 

2 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7from __future__ import annotations 

8 

9from typing import List, Union 

10 

11from oc_ocdm.support.reporter import Reporter 

12from rdflib import Dataset, Graph, Literal, URIRef 

13from SPARQLWrapper import JSON, POST, XML, SPARQLWrapper 

14 

15from rdflib_ocdm.ocdm_graph import OCDMDataset, OCDMGraph 

16from rdflib_ocdm.retry_utils import execute_with_retry 

17 

18 

19class Reader: 

20 def __init__(self, repok: Reporter | None = None, reperr: Reporter | None = None): 

21 if repok is None: 

22 self.repok: Reporter = Reporter(prefix="[Reader: INFO] ") 

23 else: 

24 self.repok: Reporter = repok 

25 

26 if reperr is None: 

27 self.reperr: Reporter = Reporter(prefix="[Reader: ERROR] ") 

28 else: 

29 self.reperr: Reporter = reperr 

30 

31 @staticmethod 

32 def import_entities_from_triplestore( 

33 ocdm_graph: Union[OCDMGraph, OCDMDataset], 

34 ts_url: str, 

35 res_list: List[URIRef], 

36 max_retries: int = 5, 

37 ) -> None: 

38 sparql: SPARQLWrapper = SPARQLWrapper(ts_url) 

39 

40 if isinstance(ocdm_graph, OCDMDataset): 

41 query: str = f""" 

42 SELECT ?g ?s ?p ?o (LANG(?o) AS ?lang) 

43 WHERE {{ 

44 GRAPH ?g {{ 

45 ?s ?p ?o. 

46 VALUES ?s {{<{"> <".join(res_list)}>}} 

47 }} 

48 }} 

49 """ 

50 sparql.setQuery(query) 

51 sparql.setMethod(POST) 

52 sparql.setReturnFormat(JSON) 

53 

54 result: dict = execute_with_retry( # type: ignore[type-arg] 

55 sparql.queryAndConvert, max_retries=max_retries 

56 ) 

57 

58 if result and "results" in result and "bindings" in result["results"]: 

59 temp_graph = Dataset() 

60 for binding in result["results"]["bindings"]: 

61 graph_uri = Graph(identifier=URIRef(binding["g"]["value"])) 

62 subject = URIRef(binding["s"]["value"]) 

63 predicate = URIRef(binding["p"]["value"]) 

64 

65 obj_data = binding["o"] 

66 if obj_data["type"] == "uri": 

67 obj = URIRef(obj_data["value"]) 

68 else: 

69 value = obj_data["value"] 

70 lang = binding.get("lang", {}).get("value") 

71 datatype = obj_data.get("datatype") 

72 

73 if lang: 

74 obj = Literal(value, lang=lang) 

75 elif datatype: 

76 obj = Literal(value, datatype=URIRef(datatype)) 

77 else: 

78 obj = Literal(value) 

79 

80 temp_graph.add((subject, predicate, obj, graph_uri)) 

81 

82 for quad in temp_graph.quads(): 

83 ocdm_graph.add(quad) # type: ignore[arg-type] 

84 else: 

85 raise ValueError("No entities were found.") 

86 

87 elif isinstance(ocdm_graph, OCDMGraph): 

88 query: str = f""" 

89 CONSTRUCT {{ 

90 ?s ?p ?o 

91 }} 

92 WHERE {{ 

93 ?s ?p ?o.  

94 VALUES ?s {{<{"> <".join(res_list)}>}} 

95 }} 

96 """ 

97 sparql.setQuery(query) 

98 sparql.setMethod(POST) 

99 sparql.setReturnFormat(XML) 

100 

101 result_graph: Graph = execute_with_retry( # type: ignore[type-arg] 

102 sparql.queryAndConvert, max_retries=max_retries 

103 ) 

104 

105 if result_graph is not None and len(result_graph) > 0: 

106 for triple in result_graph: 

107 ocdm_graph.add(triple) 

108 else: 

109 raise ValueError("No entities were found.") 

110 

111 else: 

112 raise TypeError("ocdm_graph must be either OCDMGraph or OCDMDataset")