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
« 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."""
3import logging
4from collections.abc import Iterable
6from rdflib import BNode, Dataset, Graph, Literal, URIRef
7from rdflib.util import from_n3
9from .shape import Shape
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
24 def _compile(self):
25 from . import CompiledShapes 1efghijbcak
27 if self._compiled is None: 1efghijbcak
28 self._compiled = CompiledShapes(self.graph, advanced=True) 1efghijbcak
29 return self._compiled 1efghijbcak
31 @property
32 def shapes(self):
33 from rdflib.namespace import SH 1efghijbca
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
44 def lookup_shape_from_node(self, node):
45 return self._node_shape_cache[node] 1bca
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
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
62 def objects(self, subject, predicate):
63 return self.graph.objects(subject, predicate) 1bca
65 def subjects(self, predicate, object):
66 return self.graph.subjects(predicate, object) 1a
68 def predicates(self, subject, object):
69 return self.graph.predicates(subject, object) 1a
71 def predicate_objects(self, subject):
72 return self.graph.predicate_objects(subject) 1a
74 def subject_objects(self, predicate):
75 return self.graph.subject_objects(predicate) 1a
77 def subject_predicates(self, object):
78 return self.graph.subject_predicates(object) 1a