Source code for rdflib.plugins.serializers.trig

Trig RDF graph serializer for RDFLib.
See <> 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

    from rdflib.graph import _ContextType, _SubjectType

__all__ = ["TrigSerializer"]

[docs]class TrigSerializer(TurtleSerializer): short_name = "trig" indentString = 4 * " "
[docs] 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)
[docs] 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 = 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)
[docs] def reset(self) -> None: super(TrigSerializer, self).reset() self._contexts: Dict[ _ContextType, Tuple[List[_SubjectType], Dict[_SubjectType, bool]], ] = {}
[docs] def serialize( self, stream: IO[bytes], base: Optional[str] = None, encoding: Optional[str] = None, spacious: Optional[bool] = None, **args, ): self.reset() = 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 is not None: self.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 = {} = 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"))