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

50 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-10-16 12:25 +0000

1"""Shape inspection backed by the native structural compiler and target evaluator.""" 

2 

3import logging 

4from collections.abc import Iterable 

5 

6from rdflib import BNode, Dataset, Graph, Literal, URIRef 

7from rdflib.util import from_n3 

8 

9from .shape import Shape 

10 

11 

12class ShapesGraph: 

13 def __init__(self, graph, debug=False, logger=None): 

14 if not isinstance(graph, Graph): 1efghijbcak

15 raise TypeError("ShapesGraph requires an RDFLib Graph or Dataset") 1a

16 self.graph = graph 1efghijbcak

17 if isinstance(graph, Dataset): 1efghijbcak

18 graph.default_union = True 1a

19 self.debug = debug 1efghijbcak

20 self.logger = logger or logging.getLogger(__name__) 1efghijbcak

21 self._compiled = None 1efghijbcak

22 self._node_shape_cache = {} 1efghijbcak

23 

24 def _compile(self): 

25 from . import CompiledShapes 1efghijbcak

26 

27 if self._compiled is None: 1efghijbcak

28 self._compiled = CompiledShapes(self.graph, advanced=True) 1efghijbcak

29 return self._compiled 1efghijbcak

30 

31 @property 

32 def shapes(self): 

33 from rdflib.namespace import SH 1efghijbca

34 

35 if not self._node_shape_cache: 1efghijbca

36 for term in self._compile()._native.shape_nodes(): 1efghijbca

37 node = from_n3(term) 1efghijbca

38 path = self.graph.value(node, SH.path) 1efghijbca

39 self._node_shape_cache[node] = Shape( 1efghijbca

40 self, node, p=path is not None, path=path, logger=self.logger 

41 ) 

42 return self._node_shape_cache.values() 1efghijbca

43 

44 def lookup_shape_from_node(self, node): 

45 return self._node_shape_cache[node] 1bca

46 

47 def targeting( 

48 self, 

49 data_graph, 

50 node: URIRef | BNode | Literal, 

51 *, 

52 skip: Iterable[URIRef | BNode] = (), 

53 ) -> list[URIRef | BNode]: 

54 """Return active node-shape identifiers targeting node, including AF targets.""" 

55 return self._compile().targeting(data_graph, node, skip=skip) 1k

56 

57 def shapes_from_uris(self, shapes_uris): 

58 if not self._node_shape_cache: 1bca

59 list(self.shapes) 1a

60 return [self._node_shape_cache[node] for node in shapes_uris] 1bca

61 

62 def objects(self, subject, predicate): 

63 return self.graph.objects(subject, predicate) 1bca

64 

65 def subjects(self, predicate, object): 

66 return self.graph.subjects(predicate, object) 1a

67 

68 def predicates(self, subject, object): 

69 return self.graph.predicates(subject, object) 1a

70 

71 def predicate_objects(self, subject): 

72 return self.graph.predicate_objects(subject) 1a

73 

74 def subject_objects(self, predicate): 

75 return self.graph.subject_objects(predicate) 1a

76 

77 def subject_predicates(self, object): 

78 return self.graph.subject_predicates(object) 1a