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

1"""Shared Python input and result conversion for the native backends.""" 

2 

3import json 

4from os import PathLike 

5from pathlib import Path 

6from typing import Self 

7 

8from rdflib import BNode, Graph, Literal, URIRef, Variable 

9from rdflib.query import Result 

10 

11from ._literal import restore 

12from ._native import NativeDataset 

13 

14 

15def _encode(value): 

16 if value is None: 

17 return None 

18 if isinstance(value, URIRef): 

19 return {"type": "uri", "value": str(value)} 

20 if isinstance(value, BNode): 

21 return {"type": "bnode", "value": str(value)} 

22 if isinstance(value, Literal): 

23 result = {"type": "literal", "value": str(value)} 

24 if value.language: 

25 result["xml:lang"] = value.language 

26 elif value.datatype: 

27 result["datatype"] = str(value.datatype) 

28 return result 

29 raise TypeError(f"Unsupported RDF term: {type(value).__name__}") 

30 

31 

32def _term(value): 

33 if value is None: 

34 return None 

35 kind = value["type"] 

36 if kind == "uri": 

37 return str.__new__(URIRef, value["value"]) 

38 if kind == "bnode": 

39 return str.__new__(BNode, value["value"]) 

40 if kind == "literal": 

41 return restore(value["value"], value.get("datatype"), value.get("xml:lang")) 

42 raise ValueError(f"Unknown native RDF term: {kind}") 

43 

44 

45class GraphOperations: 

46 """RDF input, output and SPARQL query operations on a native handle.""" 

47 

48 def __len__(self) -> int: 

49 return len(self._native) 

50 

51 def parse( 

52 self, 

53 source: str | PathLike[str] | None = None, 

54 *, 

55 data: str | bytes | None = None, 

56 format: str = "turtle", 

57 publicID: str | None = None, 

58 ) -> Self: 

59 """Append a UTF-8 document atomically. source is a local file path.""" 

60 if (source is None) == (data is None): 

61 raise TypeError("Specify exactly one of source or data") 

62 if source is not None: 

63 path = Path(source).resolve() 

64 data = path.read_text(encoding="utf-8") 

65 if publicID is None: 

66 publicID = path.as_uri() 

67 if isinstance(data, bytes): 

68 data = data.decode("utf-8") 

69 if not isinstance(data, str): 

70 raise TypeError("RDF data must be str or UTF-8 bytes") 

71 self._native.load(data, format, publicID) 

72 return self 

73 

74 def serialize(self, *, format: str = "nquads") -> str: 

75 """Serialize natively; graph-only formats reject named graph data.""" 

76 return self._native.dump(format) 

77 

78 def query(self, query: str, *, default_union: bool = False) -> Result: 

79 """Execute SELECT, ASK, CONSTRUCT or DESCRIBE; return an RDFLib Result.""" 

80 kind, payload = self._native.query(query, default_union) 

81 result = Result(kind) 

82 if kind == "ASK": 

83 result.askAnswer = json.loads(payload) 

84 elif kind == "SELECT": 

85 value = json.loads(payload) 

86 result.vars = [Variable(name) for name in value["variables"]] 

87 result.bindings = [ 

88 {Variable(name): _term(term) for name, term in row.items()} for row in value["rows"] 

89 ] 

90 else: 

91 native = NativeDataset() 

92 native.load(payload, "nquads", None, True) 

93 result.graph = Graph() 

94 for subject, predicate, obj, _ in json.loads(native.quads_json()): 

95 result.graph.add((_term(subject), _term(predicate), _term(obj))) 

96 return result