Coverage for triplelite / _graph.py: 88%

1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7from typing import Iterator 

8 

9from triplelite._core import TripleLite as _CTripleLite 

10from triplelite._rdflib_bridge import to_rdflib as _to_rdflib 

11from triplelite._types import RDFTerm, Triple 

12 

13 

14class SubgraphView: 

15 __slots__ = ("_parent", "_subject") 

16 

17 def __init__(self, parent: TripleLite, subject: str) -> None: 

18 self._parent = parent 

19 self._subject = subject 

20 

21 def predicate_objects(self, subject: str | None = None) -> Iterator[tuple[str, RDFTerm]]: 

22 if subject is not None and subject != self._subject: 

23 return 

24 yield from self._parent.predicate_objects(subject=self._subject) 

25 

26 def __iter__(self) -> Iterator[Triple]: 

27 yield from self._parent.triples((self._subject, None, None)) 

28 

29 def __len__(self) -> int: 

30 return sum(1 for _ in self) 

31 

32 def __eq__(self, other: object) -> bool: 

33 if isinstance(other, (set, frozenset)): 

34 return set(self) == other 

35 return NotImplemented 

36 

37 def __sub__(self, other: set | frozenset) -> set: 

38 return set(self) - other 

39 

40 def __rsub__(self, other: set | frozenset) -> set | frozenset: 

41 return other - set(self) 

42 

43 

44def _rebuild_triplelite( 

45 identifier: str | None, 

46 reverse_index_predicates: frozenset[str] | None, 

47 triples: list[Triple], 

48) -> TripleLite: 

49 g = TripleLite(identifier=identifier, reverse_index_predicates=reverse_index_predicates) 

50 g.add_many(triples) 

51 return g 

52 

53 

54class TripleLite(_CTripleLite): 

55 def __init__(self, identifier: str | None = None, reverse_index_predicates: frozenset[str] | None = None) -> None: 

56 super().__init__(identifier=identifier, reverse_index_predicates=reverse_index_predicates) 

57 self._reverse_index_predicates = reverse_index_predicates 

58 

59 def __reduce__(self) -> tuple: 

60 return (_rebuild_triplelite, (self.identifier, self._reverse_index_predicates, list(self))) 

61 

62 def subgraph(self, subject: str) -> SubgraphView | None: 

63 if not self.has_subject(subject): 

64 return None 

65 return SubgraphView(self, subject) 

66 

67 def to_rdflib(self): 

68 return _to_rdflib(self)