Coverage for python/pyshacl_turbo/target_data.py: 100.0%

1"""Explicit immutable data snapshots for repeated shape targeting.""" 

2 

3from rdflib import BNode, Literal, URIRef (empty)

4 

5from . import _native (empty)

6from .rdf import decode, input_graph (empty)

7 

8 

9class TargetData: (empty)

10 """One data snapshot, reusable across shapes and focus-node calls. 

11 

12 Changes to the original graph are visible only after preparing a new snapshot. 

13 Cached target sets belong to this snapshot and the immutable compiled shapes. 

14 """ 

15 

16 def __init__(self, data_graph): (empty)

17 data = input_graph(data_graph) 21 ctx1jklmnopqrabcdefgstvhi

18 self._native = _native.TargetData(data.source, data.format, data.base) 21 ctx1jklmnopqrabcdefgstvhi

19 self._blank_nodes = data.blank_nodes 21 ctx1jklmnopqrabcdefgstvhi

20 self._results = {} 21 ctx1jklmnopqrabcdefgstvhi

21 self._targeting_indexes = {} 21 ctx1jklmnopqrabcdefgstvhi

22 

23 def _focus_nodes(self, compiled, node, advanced): (empty)

24 key = (compiled, node, advanced) 9 ctx1jklmnopqr

25 if key not in self._results: 9 ctx1jklmnopqr

26 output = compiled._native.focus_nodes(node.n3(), self._native, advanced) 9 ctx1jklmnopqr

27 result = decode(output, compiled._blank_nodes | self._blank_nodes, preserve_ids=True) 9 ctx1jklmnopqr

28 self._results[key] = frozenset(result.objects()) 9 ctx1jklmnopqr

29 return set(self._results[key]) 9 ctx1jklmnopqr

30 

31 def _targeting(self, compiled, node, skip): (empty)

32 if not isinstance(node, (URIRef, BNode, Literal)): 12 ctx1abcdefgstvhi

33 raise TypeError("node must be an RDFLib IRI, blank node, or literal") 1 ctx1v

34 if isinstance(skip, (str, BNode, Literal)): 11 ctx1abcdefgsthi

35 raise TypeError("skip must be an iterable of shape IRIs or blank nodes") 1 ctx1t

36 skip = frozenset(skip) 10 ctx1abcdefgshi

37 if any(not isinstance(shape, (URIRef, BNode)) for shape in skip): 10 ctx1abcdefgshi

38 raise TypeError("skip must contain only shape IRIs or blank nodes") 1 ctx1s

39 key = (compiled, skip) 9 ctx1abcdefghi

40 if key not in self._targeting_indexes: 9 ctx1abcdefghi

41 output = compiled._native.targeting_index( 9 ctx1abcdefghi

42 self._native, sorted(shape.n3() for shape in skip) 

43 ) 

44 result = decode(output, compiled._blank_nodes | self._blank_nodes, preserve_ids=True) 9 ctx1abcdefghi

45 index = {} 9 ctx1abcdefghi

46 for shape, _, focus in result: 9 ctx1abcdefghi

47 index.setdefault(focus, set()).add(shape) 9 ctx1abcdefghi

48 self._targeting_indexes[key] = { 9 ctx1abcdefghi

49 focus: tuple(sorted(shapes, key=lambda shape: shape.n3())) 

50 for focus, shapes in index.items() 

51 } 

52 return list(self._targeting_indexes[key].get(node, ())) 9 ctx1abcdefghi

53 

54 

55def prepare_target_data(data_graph): (empty)

56 """Snapshot data once for repeated ``Shape.focus_nodes`` calls. 

57 

58 Preparing an existing TargetData returns it unchanged. Prepare the original 

59 graph again when its mutations should become visible. 

60 """ 

61 return data_graph if isinstance(data_graph, TargetData) else TargetData(data_graph) 21 ctx1jklmnopqrabcdefgstvhi