Coverage for target/native-coverage/profiles/8cb89e7529e5488ca16399c0daa3ee58/project/.venv/lib/python3.13/site-packages/rdflib_turbo/__init__.py: 100%

1"""Native RDF datasets. This alpha exposes a subset of the RDFLib API.""" 

2 

3import json 

4from collections.abc import Iterator 

5from typing import Self 

6 

7from rdflib import Dataset as RDFLibDataset 

8from rdflib import Graph as RDFLibGraph 

9 

10from ._native import NativeDataset 

11from ._operations import GraphOperations, _term 

12from .graph import Graph, StoreDataset 

13from .store import Store 

14 

15__version__ = "0.2.0a1" 

16__all__ = ["Dataset", "Graph", "Store", "StoreDataset"] 

17 

18 

19class Dataset(GraphOperations): 

20 """An exact-term native dataset with explicit RDFLib conversion. 

21 

22 Graph mutation is available through parse(). Inference uses staged commits. 

23 Queries use the default graph unless default_union=True is explicitly passed. 

24 """ 

25 

26 def __init__(self) -> None: 

27 self._native = NativeDataset() 

28 

29 def copy(self) -> "Dataset": 

30 result = Dataset() 

31 result._native = self._native.copy() 

32 return result 

33 

34 def namespaces(self) -> Iterator[tuple[str, str]]: 

35 """Return namespace bindings retained atomically from parsed documents.""" 

36 return iter(self._native.namespaces()) 

37 

38 def infer(self, *, inference: str = "owlrl", destination: str | None = None) -> Self: 

39 """Expand this native dataset atomically; used by owlrl-turbo.""" 

40 self._native.infer(inference, destination) 

41 return self 

42 

43 def infer_delta(self, *, inference: str = "owlrl", destination: str | None = None) -> "Dataset": 

44 """Expand atomically and return a native dataset of newly added quads. 

45 

46 The result retains namespace bindings and exact output graph contexts. 

47 It excludes original quads; failures leave this dataset unchanged. 

48 """ 

49 result = Dataset() 

50 result._native = self._native.infer_delta(inference, destination) 

51 return result 

52 

53 @classmethod 

54 def from_rdflib(cls, graph: RDFLibGraph) -> Self: 

55 """Copy an RDFLib Graph or Dataset, preserving terms and blank-node IDs.""" 

56 if getattr(graph, "_turbo_native_graph", False): 

57 raise NotImplementedError("Native Store inference/SHACL adapters are not implemented") 

58 if not isinstance(graph, RDFLibGraph): 

59 raise TypeError("Expected an RDFLib Graph or Dataset") 

60 result = cls() 

61 if isinstance(graph, RDFLibDataset): 

62 source = graph.serialize(format="nquads") 

63 rdf_format = "nquads" 

64 elif graph.context_aware: 

65 raise NotImplementedError("Convert context-aware graphs to rdflib.Dataset first") 

66 else: 

67 source = graph.serialize(format="nt") 

68 rdf_format = "nt" 

69 result._native.load(source, rdf_format, None, True) 

70 result._native.bind_namespaces( 

71 [(prefix, str(namespace)) for prefix, namespace in graph.namespaces()] 

72 ) 

73 return result 

74 

75 def to_rdflib(self) -> RDFLibDataset: 

76 """Copy to an RDFLib Dataset without literal normalization.""" 

77 result = RDFLibDataset() 

78 for prefix, namespace in self.namespaces(): 

79 result.bind(prefix, namespace) 

80 for s, p, o, g in json.loads(self._native.quads_json()): 

81 target = result.default_graph if g is None else result.graph(_term(g)) 

82 target.add((_term(s), _term(p), _term(o))) 

83 return result