Coverage for triplelite / _rdflib_bridge.py: 100%

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 TYPE_CHECKING 

8 

9from rdflib import BNode, Dataset, Graph, Literal, Node, URIRef 

10 

11from triplelite._types import _XSD_STRING, RDFTerm 

12 

13if TYPE_CHECKING: 

14 from triplelite._graph import TripleLite 

15 

16 

17def _rdflib_to_rdfterm(node: URIRef | Literal | Node | RDFTerm) -> RDFTerm: 

18 if isinstance(node, RDFTerm): 

19 return node 

20 if isinstance(node, Literal): 

21 language = str(node.language) if node.language else "" 

22 datatype = "" if language else (str(node.datatype) if node.datatype else _XSD_STRING) 

23 return RDFTerm("literal", str(node), datatype, language) 

24 return RDFTerm("uri", str(node)) 

25 

26 

27def _to_rdflib_triple(obj: RDFTerm) -> URIRef | Literal: 

28 if obj.type == "literal": 

29 if obj.lang: 

30 return Literal(obj.value, lang=obj.lang) 

31 return Literal(obj.value, datatype=URIRef(obj.datatype)) 

32 return URIRef(obj.value) 

33 

34 

35def _from_rdflib_graph(graph: Graph) -> TripleLite: 

36 from triplelite._graph import TripleLite 

37 

38 identifier = str(graph.identifier) if not isinstance(graph.identifier, BNode) else None 

39 tl = TripleLite(identifier=identifier) 

40 tl.add_many( 

41 (str(subject), str(predicate), _rdflib_to_rdfterm(obj)) 

42 for subject, predicate, obj in graph 

43 ) 

44 return tl 

45 

46 

47def from_rdflib(source: Graph | Dataset) -> list[TripleLite]: 

48 if isinstance(source, Dataset): 

49 return [_from_rdflib_graph(graph) for graph in source.graphs()] 

50 return [_from_rdflib_graph(source)] 

51 

52 

53def to_rdflib(source: TripleLite) -> Graph | Dataset: 

54 if source.identifier is not None: 

55 dataset = Dataset() 

56 named_graph = dataset.graph(URIRef(source.identifier)) 

57 for s, p, o in source: 

58 named_graph.add((URIRef(s), URIRef(p), _to_rdflib_triple(o))) 

59 return dataset 

60 graph = Graph() 

61 for s, p, o in source: 

62 graph.add((URIRef(s), URIRef(p), _to_rdflib_triple(o))) 

63 return graph