Coverage for src/metador_core/rdf/lib.py: 0%

1"""Wrappers and helpers for RDFlib.""" 

2from typing import Any, Dict, Union 

3 

4import rdflib 

5import wrapt 

6from rdflib.term import Identifier 

7 

8# NOTE: could evaluate oxrdflib if rdflib turns out being too slow 

9 

10 

11def at_most_one(g): 

12 """Return at most one element from a generator. 

13 

14 If there are more, will raise ValueError. 

15 """ 

16 ret = next(g, None) 

17 if next(g, None) is None: 

18 return ret 

19 else: 

20 raise ValueError() 

21 

22 

23class GraphNode(wrapt.ObjectProxy): 

24 """Wrapper for rdflib helping to navigate entities in an `rdflib.Graph`.""" 

25 

26 _self_graph: rdflib.Graph 

27 obj: Identifier 

28 

29 def __init__(self, graph: rdflib.Graph, obj: Identifier = None): 

30 super().__init__(obj) 

31 self._self_graph = graph 

32 

33 @property 

34 def graph(self): 

35 return self._self_graph 

36 

37 @property 

38 def node(self): 

39 return self.__wrapped__ 

40 

41 def wrap(self, obj: Identifier): 

42 return GraphNode(self.graph, obj) 

43 

44 def is_literal(self): 

45 return isinstance(self.__wrapped__, rdflib.Literal) 

46 

47 def is_uriref(self): 

48 return isinstance(self.__wrapped__, rdflib.URIRef) 

49 

50 # ---- 

51 

52 def edges_in(self): 

53 return map(self.wrap, self.graph.predicates(object=self.node, unique=True)) # type: ignore 

54 

55 def edges_out(self): 

56 return map(self.wrap, self.graph.predicates(subject=self.node, unique=True)) # type: ignore 

57 

58 def objects(self, pred): 

59 return map(self.wrap, self.graph.objects(self.node, pred, unique=True)) # type: ignore 

60 

61 def subjects(self, pred): 

62 return map(self.wrap, self.graph.subjects(pred, self.node, unique=True)) # type: ignore 

63 

64 def subject(self, pred): 

65 try: 

66 return at_most_one(self.subjects(pred)) 

67 except ValueError: 

68 msg = f"Expected to get exactly one match for: (*, {pred}, {self.obj})" 

69 raise ValueError(msg) 

70 

71 def object(self, pred): 

72 try: 

73 return at_most_one(self.objects(pred)) 

74 except ValueError: 

75 msg = f"Expected to get exactly one match for: ({self.obj}, {pred}, *)" 

76 raise ValueError(msg) 

77 

78 

79class RDFParser(wrapt.ObjectProxy): 

80 """Helper wrapper to access entity properties backed by an RDFlib graph. 

81 

82 Ensures that queries are only performed when needed (more efficient). 

83 

84 Use this e.g. to parse linked data with expected structure into your custom 

85 data structures, e.g. for use in schemas. 

86 """ 

87 

88 __wrapped__: GraphNode 

89 

90 def __init__(self, node: GraphNode): 

91 super().__init__(node) 

92 self._self_parsed: Dict[str, Any] = {} 

93 

94 def __getattr__(self, key: str): 

95 # special case: return the method 

96 if key.startswith("parse_"): 

97 return wrapt.ObjectProxy.__getattr__(self, key) 

98 

99 # try to return the attribute 

100 if val := self._self_parsed.get(key): 

101 return val 

102 # need first to compute the attribute 

103 if pfunc := getattr(self, f"parse_{key}"): 

104 val = pfunc(self.__wrapped__) 

105 self._self_parsed[key] = val 

106 return val 

107 # no parser defined -> pass through 

108 return getattr(self.__wrapped__, key) 

109 

110 @classmethod 

111 def from_graph(cls, g: rdflib.Graph, obj: Union[str, rdflib.URIRef]): 

112 return cls(GraphNode(g, rdflib.URIRef(obj)))