Skip to content

trig

Trig RDF graph serializer for RDFLib. See http://www.w3.org/TR/trig/ for syntax specification.

Classes:

__all__ module-attribute 

__all__ = ['TrigSerializer']

TrigSerializer 

TrigSerializer(store: Union[Graph, ConjunctiveGraph])

Bases: TurtleSerializer

TriG RDF graph serializer.

Methods:

Attributes:

Source code in rdflib/plugins/serializers/trig.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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)

contexts instance-attribute 

contexts = list(contexts())

default_context instance-attribute 

default_context: Optional[Node]

indentString class-attribute instance-attribute 

indentString = 4 * ' '

short_name class-attribute instance-attribute 

short_name = 'trig'

preprocess 

preprocess() -> None
Source code in rdflib/plugins/serializers/trig.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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
        # Don't generate a new prefix for a graph URI if one already exists
        self.getQName(context.identifier, False)
        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)

reset 

reset() -> None
Source code in rdflib/plugins/serializers/trig.py 
59
60
61
62
63
64
def reset(self) -> None:
    super(TrigSerializer, self).reset()
    self._contexts: Dict[
        _ContextType,
        Tuple[List[_SubjectType], Dict[_SubjectType, bool]],
    ] = {}

serialize 

serialize(stream: IO[bytes], base: Optional[str] = None, encoding: Optional[str] = None, spacious: Optional[bool] = None, **kwargs: Any) -> None
Source code in rdflib/plugins/serializers/trig.py 
 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
119
120
121
122
123
def serialize(
    self,
    stream: IO[bytes],
    base: Optional[str] = None,
    encoding: Optional[str] = None,
    spacious: Optional[bool] = None,
    **kwargs: Any,
) -> None:
    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:
                # Show the full graph URI if a prefix for it doesn't already exist
                iri = self.getQName(store.identifier, False)
                if iri is None:
                    iri = store.identifier.n3()
            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"))