Skip to content

# `` trig

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

Classes:

- `TrigSerializer` –

  TriG RDF graph serializer.

## `` __all__ `module-attribute`

    __all__ = ['TrigSerializer']

## `` TrigSerializer

    TrigSerializer(store: Union[Graph, ConjunctiveGraph])

Bases: `TurtleSerializer`

TriG RDF graph serializer.

Methods:

- `preprocess` –
- `reset` –
- `serialize` –

Attributes:

- `contexts` –
- `default_context` (`Optional[Node]`) –
- `indentString` –
- `short_name` –

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.get_pname(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.get_pname(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"))