1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
"""
Trig RDF graph serializer for RDFLib.
See <http://www.w3.org/TR/trig/> for syntax specification.
"""
from typing import IO, TYPE_CHECKING, Dict, List, Optional, Tuple, Union
from rdflib.graph import ConjunctiveGraph, Graph
from rdflib.plugins.serializers.turtle import TurtleSerializer
from rdflib.term import BNode, Node
if TYPE_CHECKING:
from rdflib.graph import _ContextType, _SubjectType
__all__ = ["TrigSerializer"]
class TrigSerializer(TurtleSerializer):
short_name = "trig"
indentString = 4 * " "
def __init__(self, store: Union[Graph, ConjunctiveGraph]):
self.default_context: Optional[Node]
if store.context_aware:
if TYPE_CHECKING:
assert isinstance(store, ConjunctiveGraph)
self.contexts = list(store.contexts())
self.default_context = store.default_context.identifier
if store.default_context:
self.contexts.append(store.default_context)
else:
self.contexts = [store]
self.default_context = None
super(TrigSerializer, self).__init__(store)
def preprocess(self) -> None:
for context in self.contexts:
# do not write unnecessary prefix (ex: for an empty default graph)
if len(context) == 0:
continue
self.store = context
self.getQName(context.identifier)
self._subjects = {}
for triple in context:
self.preprocessTriple(triple)
for subject in self._subjects.keys():
self._references[subject] += 1
self._contexts[context] = (self.orderSubjects(), self._subjects)
def reset(self) -> None:
super(TrigSerializer, self).reset()
self._contexts: Dict[
_ContextType,
Tuple[List[_SubjectType], Dict[_SubjectType, bool]],
] = {}
def serialize(
self,
stream: IO[bytes],
base: Optional[str] = None,
encoding: Optional[str] = None,
spacious: Optional[bool] = None,
**args,
):
self.reset()
self.stream = stream
# if base is given here, use that, if not and a base is set for the graph use that
if base is not None:
self.base = base
elif self.store.base is not None:
self.base = self.store.base
if spacious is not None:
self._spacious = spacious
self.preprocess()
self.startDocument()
firstTime = True
for store, (ordered_subjects, subjects) in self._contexts.items():
if not ordered_subjects:
continue
self._serialized = {}
self.store = store
self._subjects = subjects
if self.default_context and store.identifier == self.default_context:
self.write(self.indent() + "\n{")
else:
iri: Optional[str]
if isinstance(store.identifier, BNode):
iri = store.identifier.n3()
else:
iri = self.getQName(store.identifier)
if iri is None:
# type error: "IdentifiedNode" has no attribute "n3"
iri = store.identifier.n3() # type: ignore[attr-defined]
self.write(self.indent() + "\n%s {" % iri)
self.depth += 1
for subject in ordered_subjects:
if self.isDone(subject):
continue
if firstTime:
firstTime = False
if self.statement(subject) and not firstTime:
self.write("\n")
self.depth -= 1
self.write("}\n")
self.endDocument()
stream.write("\n".encode("latin-1"))
|
